Files
aios/active-runtime/app/task_plane/agent_registry.py
T

28 lines
736 B
Python

"""Agent registry abstraction."""
from __future__ import annotations
from .errors import NotFoundError
class AgentRegistry:
def __init__(self):
self._agents: dict[str, dict] = {}
def register(self, agent_id: str, metadata: dict) -> None:
self._agents[agent_id] = {"agent_id": agent_id, **metadata}
def unregister(self, agent_id: str) -> None:
if agent_id not in self._agents:
raise NotFoundError("agent")
del self._agents[agent_id]
def list(self) -> list[dict]:
return list(self._agents.values())
def get(self, agent_id: str) -> dict:
try:
return self._agents[agent_id]
except KeyError:
raise NotFoundError("agent")