Skip to content

symfonic.core.agents.in_memory

in_memory

In-memory agent store — reference implementation for testing/CI.

InMemoryAgentStore 

InMemoryAgentStore()

Reference AgentStore for testing/CI.

Source code in src/symfonic/core/agents/in_memory.py 
16
17
def __init__(self) -> None:
    self._agents: dict[str, AgentDefinition] = {}

list async 

list() -> list[AgentDefinition]

Return all stored agent definitions.

Source code in src/symfonic/core/agents/in_memory.py 
19
20
21
async def list(self) -> list[AgentDefinition]:
    """Return all stored agent definitions."""
    return list(self._agents.values())

read async 

read(name: str) -> AgentDefinition | None

Return an agent definition by name, or None if not found.

Source code in src/symfonic/core/agents/in_memory.py 
23
24
25
async def read(self, name: str) -> AgentDefinition | None:
    """Return an agent definition by name, or None if not found."""
    return self._agents.get(name)

register 

register(agent: AgentDefinition) -> None

Register an agent definition for testing. Not part of the protocol.

Source code in src/symfonic/core/agents/in_memory.py 
27
28
29
def register(self, agent: AgentDefinition) -> None:
    """Register an agent definition for testing. Not part of the protocol."""
    self._agents[agent.name] = agent

run async 

run(name: str, agent_input: Any) -> Any

Run an agent by name. Raises NotFoundError if not registered.

Source code in src/symfonic/core/agents/in_memory.py 
31
32
33
34
35
async def run(self, name: str, agent_input: Any) -> Any:
    """Run an agent by name. Raises NotFoundError if not registered."""
    if name not in self._agents:
        raise NotFoundError(f"Agent '{name}' not found")
    return {"agent": name, "input": agent_input, "result": "stub"}