Delegation MCP

"""Main Gradio application with monitoring and configuration tabs.""" 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 from .config_tab import create_config_tab from .config_manager import ConfigurationManager 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_app( config_manager: ConfigurationManager | None = None, db_path: Path = Path("data/delegation.db"), ): """Create main Gradio application with multiple tabs. Args: config_manager: Optional ConfigurationManager instance db_path: Path to delegation database for monitoring Returns: Gradio Blocks application with Monitor and Configuration tabs """ if not GRADIO_AVAILABLE: print("Error: Gradio is not installed. Please install with `pip install .[ui]`") return None if config_manager is None: config_manager = ConfigurationManager() monitor = DelegationMonitor(db_path) # Create the main application with tabs with gr.Blocks( title="Delegation MCP - Monitor & Configuration", theme=gr.themes.Soft(), ) as app: gr.Markdown(""" # 🚀 Delegation MCP - Multi-Agent Orchestration **Monitor delegation activity and configure agent routing in real-time.** This interface provides two main functions: - **Monitor**: View live delegation activity and statistics (for demos and debugging) - **Configuration**: Manage agents and routing rules with immediate effect """) with gr.Tabs() as tabs: # Tab 1: Monitor with gr.Tab("📊 Monitor"): gr.Markdown(""" # 🔍 Delegation MCP - Live Activity Monitor **This monitor shows real-time delegation activity** when Claude Code (or other MCP clients) call the delegation MCP server. **How to use:** 1. Start the MCP server: `delegation-mcp` 2. Configure Claude Code to use it (see README.md) 3. Chat with Claude Code and ask it to delegate tasks 4. Watch delegations appear here in real-time! --- """) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### 📊 Statistics") total_tasks = gr.Number(label="Total Tasks", value=0, interactive=False) success_rate = gr.Number(label="Success Rate (%)", value=0, interactive=False) avg_duration = gr.Number(label="Avg Duration (s)", value=0, interactive=False) with gr.Column(scale=2): gr.Markdown("### 🤖 Agent Usage") agent_chart = gr.BarPlot( x="agent", y="count", title="Delegations by Agent", ) gr.Markdown("### 📝 Recent Delegations") activity_table = gr.Dataframe( headers=["Time", "From", "To", "Status", "Duration"], label="Live Activity", interactive=False, wrap=True, ) refresh_btn = gr.Button("🔄 Refresh", variant="primary") def refresh_all(): """Refresh all monitor data.""" # Get statistics stats = monitor.get_statistics() # Get recent activity activity = monitor.get_recent_activity() # Format agent usage for chart agent_data = [] for agent, count in stats["agent_usage"].items(): agent_data.append({"agent": agent, "count": count}) return ( stats["total"], stats["success_rate"] * 100, stats["avg_duration"], {"data": agent_data} if agent_data else None, activity, ) # Wire up refresh button refresh_btn.click( fn=refresh_all, outputs=[total_tasks, success_rate, avg_duration, agent_chart, activity_table], ) # Auto-refresh on load app.load( fn=refresh_all, outputs=[total_tasks, success_rate, avg_duration, agent_chart, activity_table], ) # Tab 2: Configuration with gr.Tab("⚙️ Configuration"): config_tab = create_config_tab(config_manager) gr.Markdown(""" --- ### Getting Started 1. Configure agents in the Configuration tab 2. Set up routing rules for automatic task delegation 3. Start the MCP server: `delegation-mcp` 4. Connect MCP clients (Claude Code, etc.) 5. Watch delegations in the Monitor tab Changes take effect immediately. See [GitHub](https://github.com/carlosduplar/multi-agent-mcp) for docs. """) return app def main( server_name: str = "0.0.0.0", server_port: int = 7860, share: bool = False, ): """Launch the Gradio application. Args: server_name: Server hostname (default: 0.0.0.0 for all interfaces) server_port: Server port (default: 7860) share: Enable Gradio share link (default: False) """ app = create_app() if app: app.launch( server_name=server_name, server_port=server_port, share=share, ) if __name__ == "__main__": main()