forked from Rockachopa/Timmy-time-dashboard
## 1. MCP (Model Context Protocol) Implementation ### Registry (src/mcp/registry.py) - Tool registration with JSON schemas - Dynamic tool discovery - Health tracking per tool - Metrics collection (latency, error rates) - @register_tool decorator for easy registration ### Server (src/mcp/server.py) - MCPServer class implementing MCP protocol - MCPHTTPServer for FastAPI integration - Standard endpoints: list_tools, call_tool, get_schema ### Schemas (src/mcp/schemas/base.py) - create_tool_schema() helper - Common parameter types - Standard return types ### Bootstrap (src/mcp/bootstrap.py) - Automatic tool module loading - Status reporting ## 2. MCP-Compliant Tools (src/tools/) | Tool | Purpose | Category | |------|---------|----------| | web_search | DuckDuckGo search | research | | read_file | File reading | files | | write_file | File writing (confirmation) | files | | list_directory | Directory listing | files | | python | Python code execution | code | | memory_search | Vector memory search | memory | All tools have proper schemas, error handling, and MCP registration. ## 3. Event Bus (src/events/bus.py) - Async publish/subscribe pattern - Pattern matching with wildcards (agent.task.*) - Event history tracking - Concurrent handler execution - Module-level singleton for system-wide use ## 4. Sub-Agents (src/agents/) All agents inherit from BaseAgent with: - Agno Agent integration - MCP tool registry access - Event bus connectivity - Structured logging ### Agent Roster | Agent | Role | Tools | Purpose | |-------|------|-------|---------| | Seer | Research | web_search, read_file, memory_search | Information gathering | | Forge | Code | python, write_file, read_file | Code generation | | Quill | Writing | write_file, read_file, memory_search | Content creation | | Echo | Memory | memory_search, read_file, write_file | Context retrieval | | Helm | Routing | memory_search | Task routing decisions | | Timmy | Orchestrator | All tools | Coordination & user interface | ### Timmy Orchestrator - Analyzes user requests - Routes to appropriate sub-agent - Handles direct queries - Manages swarm coordination - create_timmy_swarm() factory function ## 5. Integration All components wired together: - Tools auto-register on import - Agents connect to event bus - MCP server provides HTTP API - Ready for dashboard integration ## Tests - All 973 existing tests pass - New components tested manually - Import verification successful Next steps: Cascade Router, Self-Upgrade Loop, Dashboard integration
107 lines
3.0 KiB
Python
107 lines
3.0 KiB
Python
"""Helm Agent — Routing and orchestration decisions.
|
|
|
|
Capabilities:
|
|
- Task analysis
|
|
- Agent selection
|
|
- Workflow planning
|
|
- Priority management
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from agents.base import BaseAgent
|
|
|
|
|
|
HELM_SYSTEM_PROMPT = """You are Helm, a routing and orchestration specialist.
|
|
|
|
Your role is to analyze tasks and decide how to route them to other agents.
|
|
|
|
## Capabilities
|
|
|
|
- Task analysis and decomposition
|
|
- Agent selection for tasks
|
|
- Workflow planning
|
|
- Priority assessment
|
|
|
|
## Guidelines
|
|
|
|
1. **Analyze carefully** — Understand what the task really needs
|
|
2. **Route wisely** — Match tasks to agent strengths
|
|
3. **Consider dependencies** — Some tasks need sequencing
|
|
4. **Be efficient** — Don't over-complicate simple tasks
|
|
|
|
## Agent Roster
|
|
|
|
- Seer: Research, information gathering
|
|
- Forge: Code, tools, system changes
|
|
- Quill: Writing, documentation
|
|
- Echo: Memory, context retrieval
|
|
- Mace: Security, validation (use for sensitive operations)
|
|
|
|
## Response Format
|
|
|
|
Provide routing decisions as:
|
|
- Task breakdown (subtasks if needed)
|
|
- Agent assignment (who does what)
|
|
- Execution order (sequence if relevant)
|
|
- Rationale (why this routing)
|
|
|
|
You work for Timmy, the sovereign AI orchestrator. Be the dispatcher that keeps everything flowing.
|
|
"""
|
|
|
|
|
|
class HelmAgent(BaseAgent):
|
|
"""Routing and orchestration specialist."""
|
|
|
|
def __init__(self, agent_id: str = "helm") -> None:
|
|
super().__init__(
|
|
agent_id=agent_id,
|
|
name="Helm",
|
|
role="routing",
|
|
system_prompt=HELM_SYSTEM_PROMPT,
|
|
tools=["memory_search"], # May need to check past routing decisions
|
|
)
|
|
|
|
async def execute_task(self, task_id: str, description: str, context: dict) -> Any:
|
|
"""Execute a routing task."""
|
|
prompt = f"Analyze and route this task:\n\nTask: {description}\n\nProvide routing decision with rationale."
|
|
|
|
result = await self.run(prompt)
|
|
|
|
return {
|
|
"task_id": task_id,
|
|
"agent": self.agent_id,
|
|
"result": result,
|
|
"status": "completed",
|
|
}
|
|
|
|
async def route_request(self, request: str) -> dict:
|
|
"""Analyze a request and suggest routing."""
|
|
prompt = f"""Analyze this request and determine the best agent(s) to handle it:
|
|
|
|
Request: {request}
|
|
|
|
Respond in this format:
|
|
Primary Agent: [agent name]
|
|
Reason: [why this agent]
|
|
Secondary Agents: [if needed]
|
|
Complexity: [simple/moderate/complex]
|
|
"""
|
|
result = await self.run(prompt)
|
|
|
|
# Parse result into structured format
|
|
# This is simplified - in production, use structured output
|
|
return {
|
|
"analysis": result,
|
|
"primary_agent": self._extract_agent(result),
|
|
}
|
|
|
|
def _extract_agent(self, text: str) -> str:
|
|
"""Extract agent name from routing text."""
|
|
agents = ["seer", "forge", "quill", "echo", "mace", "helm"]
|
|
text_lower = text.lower()
|
|
for agent in agents:
|
|
if agent in text_lower:
|
|
return agent
|
|
return "timmy" # Default to orchestrator
|