File size: 8,459 Bytes
8bab08d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 |
"""
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
)
|