app_mcp_autonomous.py

"""
CX AI Agent - Autonomous MCP Demo

This is the PROPER MCP implementation where:
- AI (Claude 3.5 Sonnet) autonomously calls MCP tools
- NO hardcoded workflow
- AI decides which tools to use and when
- Full Model Context Protocol demonstration

Perfect for MCP hackathon!
"""

import os
import gradio as gr
import asyncio
from pathlib import Path
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Set in-memory MCP mode for HF Spaces
os.environ["USE_IN_MEMORY_MCP"] = "true"

from mcp.registry import get_mcp_registry
from mcp.agents.autonomous_agent import AutonomousMCPAgent


# Initialize MCP registry
mcp_registry = get_mcp_registry()


async def run_autonomous_agent(task: str, api_key: str):
    """
    Run the autonomous AI agent with MCP tool calling.

    Args:
        task: The task for the AI to complete autonomously
        api_key: Anthropic API key for Claude

    Yields:
        Progress updates from the agent
    """

    if not api_key:
        yield "❌ Error: Please provide an Anthropic API key"
        return

    if not task:
        yield "❌ Error: Please provide a task description"
        return

    # Create autonomous agent
    try:
        agent = AutonomousMCPAgent(mcp_registry=mcp_registry, api_key=api_key)
    except Exception as e:
        yield f"❌ Error initializing agent: {str(e)}"
        return

    # Run agent autonomously
    output_text = ""

    try:
        async for event in agent.run(task, max_iterations=15):
            event_type = event.get("type")
            message = event.get("message", "")

            # Format the message based on event type
            if event_type == "agent_start":
                output_text += f"\n{'='*60}\n"
                output_text += f"{message}\n"
                output_text += f"Model: {event.get('model')}\n"
                output_text += f"{'='*60}\n\n"

            elif event_type == "iteration_start":
                output_text += f"\n{message}\n"

            elif event_type == "tool_call":
                tool = event.get("tool")
                tool_input = event.get("input", {})
                output_text += f"\n{message}\n"
                output_text += f"  Input: {tool_input}\n"

            elif event_type == "tool_result":
                tool = event.get("tool")
                result = event.get("result", {})
                output_text += f"{message}\n"

                # Show some result details
                if isinstance(result, dict):
                    if "count" in result:
                        output_text += f"  → Returned {result['count']} items\n"
                    elif "status" in result:
                        output_text += f"  → Status: {result['status']}\n"

            elif event_type == "tool_error":
                tool = event.get("tool")
                error = event.get("error")
                output_text += f"\n{message}\n"
                output_text += f"  Error: {error}\n"

            elif event_type == "agent_complete":
                final_response = event.get("final_response", "")
                iterations = event.get("iterations", 0)
                output_text += f"\n{'='*60}\n"
                output_text += f"{message}\n"
                output_text += f"Iterations: {iterations}\n"
                output_text += f"{'='*60}\n\n"
                output_text += f"**Final Response:**\n\n{final_response}\n"

            elif event_type == "agent_error":
                error = event.get("error")
                output_text += f"\n{message}\n"
                output_text += f"Error: {error}\n"

            elif event_type == "agent_max_iterations":
                iterations = event.get("iterations", 0)
                output_text += f"\n{message}\n"

            yield output_text

    except Exception as e:
        output_text += f"\n\n❌ Agent execution failed: {str(e)}\n"
        yield output_text


def create_demo():
    """Create Gradio demo interface"""

    with gr.Blocks(title="CX AI Agent - Autonomous MCP Demo", theme=gr.themes.Soft()) as demo:
        gr.Markdown("""
        # 🤖 CX AI Agent - Autonomous MCP Demo

        This demo shows **true AI-driven MCP usage** where Claude 3.5 Sonnet:
        - ✅ Autonomously decides which MCP tools to call
        - ✅ Uses Model Context Protocol servers (Search, Store, Email, Calendar)
        - ✅ NO hardcoded workflow - AI makes all decisions
        - ✅ Proper MCP protocol implementation

        ## Available MCP Tools:
        - 🔍 **Search**: Web search, news search
        - 💾 **Store**: Save/retrieve prospects, companies, contacts, facts
        - 📧 **Email**: Send emails, track threads
        - 📅 **Calendar**: Suggest meeting times, generate invites

        ## Example Tasks:
        - "Research Shopify and determine if they're a good B2B prospect"
        - "Find 3 e-commerce companies and save them as prospects"
        - "Create a personalized outreach campaign for Stripe"
        - "Find recent news about AI startups and save as facts"
        """)

        with gr.Row():
            with gr.Column():
                api_key_input = gr.Textbox(
                    label="Anthropic API Key",
                    type="password",
                    placeholder="sk-ant-...",
                    info="Required for Claude 3.5 Sonnet (get one at console.anthropic.com)"
                )

                task_input = gr.Textbox(
                    label="Task for AI Agent",
                    placeholder="Research Shopify and create a prospect profile with facts",
                    lines=3,
                    info="Describe what you want the AI to do autonomously"
                )

                # Example tasks dropdown
                example_tasks = gr.Dropdown(
                    label="Example Tasks (click to use)",
                    choices=[
                        "Research Shopify and determine if they're a good B2B SaaS prospect",
                        "Find recent news about Stripe and save as facts in the database",
                        "Create a prospect profile for Notion including company info and facts",
                        "Search for B2B SaaS companies in the e-commerce space and save top 3 prospects",
                        "Research Figma's recent product launches and save relevant facts",
                    ],
                    interactive=True
                )

                def use_example(example):
                    return example

                example_tasks.change(fn=use_example, inputs=[example_tasks], outputs=[task_input])

                run_btn = gr.Button("🚀 Run Autonomous Agent", variant="primary", size="lg")

            with gr.Column():
                output = gr.Textbox(
                    label="Agent Progress & Results",
                    lines=25,
                    max_lines=50,
                    show_copy_button=True
                )

        run_btn.click(
            fn=run_autonomous_agent,
            inputs=[task_input, api_key_input],
            outputs=[output]
        )

        gr.Markdown("""
        ## 🎯 How It Works

        1. **You provide a task** - Tell the AI what you want to accomplish
        2. **AI analyzes the task** - Claude understands what needs to be done
        3. **AI decides which tools to use** - Autonomously chooses MCP tools
        4. **AI executes tools** - Calls MCP servers (search, store, email, calendar)
        5. **AI continues until complete** - Keeps working until task is done

        ## 🏆 True MCP Implementation

        This is **NOT** a hardcoded workflow! The AI:
        - ✅ Decides which tools to call based on context
        - ✅ Adapts to new information
        - ✅ Can call tools in any order
        - ✅ Reasons about what information it needs
        - ✅ Stores data for later use

        ## 💡 Tips

        - Be specific about what you want
        - The AI can search, save data, and reason about prospects
        - Try multi-step tasks to see autonomous decision-making
        - Check the progress log to see which tools the AI chooses

        ---

        **Powered by:** Claude 3.5 Sonnet + Model Context Protocol (MCP)
        """)

    return demo


if __name__ == "__main__":
    demo = create_demo()
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True
    )