Delegation MCP

""" Minimal Gradio monitor for demo videos. This is NOT a chat interface - it's a live activity monitor that shows delegation events happening when Claude Code (or other MCP clients) call the delegation MCP server. Purpose: Makes demo videos visually compelling by showing real-time delegation activity. """ try: import gradio as gr GRADIO_AVAILABLE = True except ImportError: GRADIO_AVAILABLE = False # Mock gr for type hinting if needed, or just handle availability check gr = None # type: ignore from pathlib import Path from datetime import datetime from collections import deque # from .persistence import PersistenceManager class DelegationMonitor: """Monitors delegation activity for demo visualization.""" def __init__(self, db_path: Path = Path("data/delegation.db")): # self.persistence = PersistenceManager(db_path) self.recent_events = deque(maxlen=20) # Keep last 20 events def get_recent_activity(self): """Get recent delegation events for display.""" return [] # try: # history = self.persistence.get_task_history(limit=20) # return [ # [ # entry.timestamp.strftime("%H:%M:%S"), # entry.orchestrator, # entry.delegated_to or "N/A", # "✅" if entry.success else "❌", # f"{entry.duration:.2f}s" # ] # for entry in history # ] # except Exception: # return [] def get_statistics(self): """Get delegation statistics for charts.""" return {"total": 0, "success_rate": 0.0, "avg_duration": 0.0, "agent_usage": {}} # try: # stats = self.persistence.get_statistics() # return { # "total": stats.get("total_tasks", 0), # "success_rate": stats.get("success_rate", 0.0), # "avg_duration": stats.get("avg_duration", 0.0), # "agent_usage": stats.get("agent_usage", {}), # } # except Exception: # return {"total": 0, "success_rate": 0.0, "avg_duration": 0.0, "agent_usage": {}} def create_monitor_ui(demo_server=None): """Create minimal monitoring UI for demo videos (< 150 lines).""" if not GRADIO_AVAILABLE: print("Error: Gradio is not installed. Please install with `pip install .[ui]`") return None monitor = DelegationMonitor() with gr.Blocks(title="Delegation MCP Monitor") as app: gr.Markdown(""" # 🚀 Delegation MCP - Interactive Demo **Test the intelligent routing system!** Enter a query below to see which agent the MCP would route it to. Adjust settings to see how configuration affects routing. --- """) # Configuration Section with gr.Accordion("⚙️ Task Routing Configuration", open=False): gr.Markdown("### 1️⃣ Select Available Agents") gr.Markdown("*Choose which AI agents are installed and available on your system*") with gr.Row(): agent_gemini = gr.Checkbox(label="🔷 Gemini", value=True) agent_claude = gr.Checkbox(label="🟣 Claude", value=True) agent_aider = gr.Checkbox(label="🟢 Aider", value=True) agent_copilot = gr.Checkbox(label="🔵 Copilot", value=False) gr.Markdown("### 2️⃣ Assign Agents to Task Categories") gr.Markdown("*Configure which agent handles each type of task - same options as install.py*") with gr.Row(): with gr.Column(): security_agent = gr.Dropdown(["gemini", "claude", "aider", "copilot"], value="gemini", label="🔒 Security Audits") architecture_agent = gr.Dropdown(["claude", "gemini", "aider", "copilot"], value="claude", label="🏗️ Architecture Design") refactoring_agent = gr.Dropdown(["aider", "claude", "gemini", "copilot"], value="aider", label="🔧 Refactoring") quick_fix_agent = gr.Dropdown(["aider", "claude", "gemini", "copilot"], value="aider", label="⚡ Quick Fixes") with gr.Column(): code_review_agent = gr.Dropdown(["claude", "gemini", "aider", "copilot"], value="claude", label="👀 Code Review") performance_agent = gr.Dropdown(["gemini", "claude", "aider", "copilot"], value="gemini", label="🚀 Performance") testing_agent = gr.Dropdown(["claude", "gemini", "aider", "copilot"], value="claude", label="🧪 Testing") git_agent = gr.Dropdown(["aider", "claude", "gemini", "copilot"], value="aider", label="📦 Git Operations") gr.Markdown(""" **Try it:** Change "Security Audits" from Gemini to Claude, then test a security query! **Note:** These are the "Balanced" preset defaults - experiment with different combinations! """) # Interactive Query Tester with gr.Row(): with gr.Column(): gr.Markdown("### 🧪 Test Routing Intelligence") query_input = gr.Textbox( label="Enter your query", placeholder="e.g., 'Scan my code for SQL injection vulnerabilities'", lines=3 ) submit_btn = gr.Button("🔍 Get Routing Decision", variant="primary", size="lg") # Example queries with gr.Accordion("💡 Example Queries", open=True): gr.Markdown("**Simple Tasks:**") gr.Examples( examples=[ ["Scan this codebase for security vulnerabilities"], ["Fix the bug causing tests to fail"], ["Create a pull request with my changes"], ], inputs=query_input, ) gr.Markdown("**Multi-Agent Workflows:**") gr.Examples( examples=[ ["Scan the codebase for frontend optimization improvements. Generate a report and fix the critical issues. Commit, push, deploy to preview, and test improvements with browser automation"], ["Audit the authentication system for SQL injection and XSS. Create detailed security report with CVE references. Fix all critical vulnerabilities and update tests"], ["Analyze API performance bottlenecks using profiling. Optimize database queries and add caching. Generate benchmark report comparing before/after metrics"], ["Review the entire codebase for code quality issues. Refactor problem areas with proper error handling. Update documentation and create comprehensive test suite"], ["Design microservices architecture for user management. Implement the service with authentication. Set up CI/CD pipeline and deploy to staging with monitoring"], ], inputs=query_input, ) with gr.Column(): gr.Markdown("### 📊 Routing Decision") decision_output = gr.Textbox( label="Decision", lines=2, interactive=False ) task_type_output = gr.Textbox( label="Task Classification", lines=1, interactive=False ) complexity_output = gr.Textbox( label="Complexity Assessment", lines=1, interactive=False ) reasoning_output = gr.Textbox( label="Routing Reasoning", lines=4, interactive=False ) cli_command_output = gr.Textbox( label="CLI Command (if delegated)", lines=2, interactive=False ) def update_agent_dropdowns(gemini_enabled, claude_enabled, aider_enabled, copilot_enabled, sec_current, arch_current, refactor_current, fix_current, review_current, perf_current, test_current, git_current): """Update dropdown choices based on selected agents and auto-reassign if needed.""" # Build list of available agents available = [] if gemini_enabled: available.append("gemini") if claude_enabled: available.append("claude") if aider_enabled: available.append("aider") if copilot_enabled: available.append("copilot") # If no agents selected, default to all if not available: available = ["gemini", "claude", "aider", "copilot"] # Helper function to get valid value (auto-reassign if current is disabled) def get_valid_value(current_value): if current_value in available: return current_value # Auto-reassign to first available agent return available[0] # Return updates for all 8 dropdowns return [ gr.update(choices=available, value=get_valid_value(sec_current)), gr.update(choices=available, value=get_valid_value(arch_current)), gr.update(choices=available, value=get_valid_value(refactor_current)), gr.update(choices=available, value=get_valid_value(fix_current)), gr.update(choices=available, value=get_valid_value(review_current)), gr.update(choices=available, value=get_valid_value(perf_current)), gr.update(choices=available, value=get_valid_value(test_current)), gr.update(choices=available, value=get_valid_value(git_current)), ] async def test_routing(query, sec_agent, arch_agent, refactor_agent, fix_agent, review_agent, perf_agent, test_agent, git_agent): """Test routing for a query without executing.""" if not query.strip(): return "Please enter a query", "", "", "", "" if not demo_server: return "❌ Server not initialized", "", "", "", "" try: # Map task categories to selected agents task_agent_map = { "security_audit": sec_agent, "vulnerability_scan": sec_agent, "architecture": arch_agent, "refactoring": refactor_agent, "quick_fix": fix_agent, "code_review": review_agent, "performance": perf_agent, "testing": test_agent, "git_workflow": git_agent, "github_operations": git_agent, } # Temporarily update routing rules based on user selection from delegation_mcp.config import DelegationRule demo_server.config.rules.clear() # Add rules for each task category for task_type, agent in task_agent_map.items(): rule = DelegationRule( delegate_to=agent, pattern=task_type, description=f"User configured: {task_type} → {agent}", priority=10 ) demo_server.config.rules.append(rule) # Get routing guidance result = await demo_server.engine.process( query, guidance_only=True ) # Classify task type and complexity task_type, timeout = demo_server.engine._classify_task(query) complexity = demo_server.engine._estimate_task_complexity(query) # Determine delegation and get reasoning target, rule = demo_server.engine._determine_delegation(query, None) # Build reasoning explanation reasoning_parts = [] # Complexity assessment reasoning_parts.append(f"📏 Complexity: {complexity.upper()}") # Task type detection query_lower = query.lower() keywords = { "security_audit": ["security", "vulnerability", "audit", "cve"], "architecture": ["architecture", "design", "system design"], "refactoring": ["refactor", "restructure", "clean up"], "quick_fix": ["fix", "bug", "error"], "performance": ["performance", "optimize", "speed"], "code_review": ["review", "code quality", "best practice"], } detected_keywords = [] for task, kws in keywords.items(): if task == task_type: detected_keywords = [kw for kw in kws if kw in query_lower] break if detected_keywords: reasoning_parts.append(f"🔍 Keywords: {', '.join(detected_keywords[:3])}") # Routing strategy if rule: if "User configured" in rule.description: reasoning_parts.append(f"⚙️ User Configuration: {task_type} → {result.delegated_to or 'claude'}") else: reasoning_parts.append(f"📋 Matched Rule: {rule.pattern}") elif complexity == "simple": reasoning_parts.append("⚡ Simple task - no delegation overhead needed") else: reasoning_parts.append(f"🎯 Capability-based routing for {task_type}") reasoning = "\n".join(reasoning_parts) # Format decision if result.delegated_to: decision = f"✅ DELEGATE TO: {result.delegated_to.upper()}" cli_cmd = f'{result.delegated_to} "{query}"' else: decision = "✅ HANDLE DIRECTLY (Claude)" cli_cmd = "N/A - Claude handles internally" task_info = f"{task_type.replace('_', ' ').title()}" complexity_info = f"{complexity.title()} (timeout: {timeout}s)" return decision, task_info, complexity_info, reasoning, cli_cmd except Exception as e: return f"❌ Error: {str(e)}", "", "", "", "" submit_btn.click( fn=test_routing, inputs=[ query_input, security_agent, architecture_agent, refactoring_agent, quick_fix_agent, code_review_agent, performance_agent, testing_agent, git_agent ], outputs=[decision_output, task_type_output, complexity_output, reasoning_output, cli_command_output] ) # Wire up checkbox changes to update dropdowns agent_checkboxes = [agent_gemini, agent_claude, agent_aider, agent_copilot] task_dropdowns = [security_agent, architecture_agent, refactoring_agent, quick_fix_agent, code_review_agent, performance_agent, testing_agent, git_agent] for checkbox in agent_checkboxes: checkbox.change( fn=update_agent_dropdowns, inputs=agent_checkboxes + task_dropdowns, outputs=task_dropdowns ) gr.Markdown("---") gr.Markdown(""" ## 🚀 Want to Test the Full System? This demo shows the **routing intelligence** only. To see actual task delegation with AI agents: ### Option 1: Duplicate This Space 1. Click the **⋮** menu (top right) → **Duplicate Space** 2. Add your API keys in **Settings → Secrets**: - `ANTHROPIC_API_KEY` - For Claude - `GOOGLE_API_KEY` - For Gemini 3. Install agent CLIs in your duplicated Space (optional for full functionality) ### Option 2: Install Locally ```bash git clone https://github.com/carlosduplar/multi-agent-mcp.git cd multi-agent-mcp pip install -e . python install.py ``` Then configure Claude Code to use the MCP server and start delegating tasks! --- ### 📚 Learn More - **GitHub**: [carlosduplar/multi-agent-mcp](https://github.com/carlosduplar/multi-agent-mcp) - **Documentation**: Full setup guide in README - **MCP Hackathon**: Built for MCP 1st Birthday (Winter 2025) """) return app def main(): """Launch monitor UI.""" app = create_monitor_ui() if app: app.launch(server_name="0.0.0.0", server_port=7860, share=False) if __name__ == "__main__": main()