Spaces:
Sleeping
Sleeping
File size: 7,578 Bytes
8b02e7c | 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 | """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()
|