112 lines
3.6 KiB
Python
112 lines
3.6 KiB
Python
from typing import Any
|
|
|
|
from app.agents.base import AgentResult, AgentStatus, BaseAgent
|
|
|
|
from app.agents.doc_agent import DocAgent
|
|
from app.agents.backlog_agent import BacklogAgent
|
|
from app.agents.spec_agent import SpecAgent
|
|
from app.agents.audit_agent import AuditAgent
|
|
from app.agents.observer_agent import ObserverAgent
|
|
from app.agents.evolution_agent import EvolutionAgent
|
|
from app.agents.security_agent import SecurityAgent
|
|
from app.agents.qa_tester_agent import QATesterAgent
|
|
from app.agents.fix_agent import FixAgent
|
|
from app.agents.ui_test_agent import UITestAgent
|
|
from app.agents.rollout_agent import RolloutAgent
|
|
from app.agents.conductor_agent import ConductorAgent
|
|
from app.agents.supervisor_agent import SupervisorAgent
|
|
from app.agents.meta_agent import MetaAgent
|
|
|
|
|
|
class AgentRegistry:
|
|
def __init__(self):
|
|
self._agents: dict[str, BaseAgent] = {}
|
|
self._initialize_agents()
|
|
|
|
def _initialize_agents(self) -> None:
|
|
self.register(DocAgent())
|
|
self.register(BacklogAgent())
|
|
self.register(SpecAgent())
|
|
self.register(AuditAgent())
|
|
self.register(ObserverAgent())
|
|
self.register(EvolutionAgent())
|
|
self.register(SecurityAgent())
|
|
self.register(QATesterAgent())
|
|
self.register(FixAgent())
|
|
self.register(UITestAgent())
|
|
self.register(RolloutAgent())
|
|
self.register(ConductorAgent())
|
|
self.register(SupervisorAgent())
|
|
self.register(MetaAgent())
|
|
|
|
def register(self, agent: BaseAgent) -> None:
|
|
if not agent.name:
|
|
raise ValueError("Agent must have a name")
|
|
self._agents[agent.name] = agent
|
|
|
|
def get(self, name: str) -> BaseAgent | None:
|
|
return self._agents.get(name)
|
|
|
|
def list_agents(self) -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"name": agent.name,
|
|
"version": agent.version,
|
|
"description": agent.description,
|
|
"status": agent.status.value,
|
|
"last_run": agent.last_run.isoformat() if agent.last_run else None,
|
|
}
|
|
for agent in self._agents.values()
|
|
]
|
|
|
|
async def run_agent(self, name: str, context: dict[str, Any] | None = None) -> AgentResult:
|
|
agent = self.get(name)
|
|
if not agent:
|
|
return AgentResult(
|
|
success=False,
|
|
message=f"Agent not found: {name}",
|
|
)
|
|
return await agent.run(context)
|
|
|
|
async def run_all(self, context: dict[str, Any] | None = None) -> dict[str, AgentResult]:
|
|
results = {}
|
|
for name, agent in self._agents.items():
|
|
try:
|
|
results[name] = await agent.run(context)
|
|
except Exception as e:
|
|
results[name] = AgentResult(
|
|
success=False,
|
|
message=f"Agent failed: {str(e)}",
|
|
errors=[str(e)],
|
|
)
|
|
return results
|
|
|
|
async def health_check_all(self) -> dict[str, bool]:
|
|
results = {}
|
|
for name, agent in self._agents.items():
|
|
try:
|
|
results[name] = await agent.health_check()
|
|
except Exception:
|
|
results[name] = False
|
|
return results
|
|
|
|
def get_metrics_all(self) -> dict[str, dict[str, Any]]:
|
|
results = {}
|
|
for name, agent in self._agents.items():
|
|
results[name] = {
|
|
"status": agent.status.value,
|
|
"version": agent.version,
|
|
}
|
|
return results
|
|
|
|
|
|
registry = AgentRegistry()
|
|
|
|
|
|
def get_agent(name: str) -> BaseAgent | None:
|
|
return registry.get(name)
|
|
|
|
|
|
def get_all_agents() -> list[dict[str, Any]]:
|
|
return registry.list_agents()
|