Skip to content

Multi-Agent

Task queue, agent registry, and agent-as-tool coordination.

TaskQueue Protocol

TaskQueue

Bases: Protocol

Async task queue for multi-agent coordination.

Exactly 5 methods (ISP limit). All methods are async.

put async 

put(item: TaskItem) -> None

Add a task to the queue.

get async 

get(filters: TaskFilter | None = None) -> TaskItem | None

Claim the highest-priority TODO task matching filters.

Implementations must return the claimed task in IN_PROGRESS status. Returns None if no matching task is available.

complete async 

complete(task_id: str) -> bool

Mark a task as done. Returns True if found and completed.

cancel async 

cancel(task_id: str) -> bool

Mark a task as cancelled. Returns True if found and cancelled.

list_tasks async 

list_tasks(filters: TaskFilter | None = None) -> list[TaskItem]

List all tasks matching filters. Returns all if filters is None.

TaskItem

TaskItem dataclass

Immutable task entry in the queue.

Attributes: id: Unique task identifier. title: Short human-readable title. description: Detailed task description. status: Current task status. priority: Task priority for scheduling. assignee_agent_id: Agent assigned to this task (None = unassigned). metadata: Arbitrary key-value metadata. created_at: Unix timestamp of creation.

InMemoryTaskQueue

InMemoryTaskQueue

In-memory task queue. Thread-safe via asyncio.Lock.

AgentRegistry Protocol

AgentRegistry

Bases: Protocol

Port: registry for managing agent lifecycle records.

Provides CRUD operations for AgentRecord instances with filtering support. Exactly 5 methods (ISP limit).

AgentRecord

AgentRecord dataclass

Immutable record describing a registered agent.

Fields: id: Unique agent identifier. name: Human-readable agent name. role: Agent's role (e.g. "researcher", "coder", "reviewer"). parent_id: ID of the parent agent (None for top-level). runtime_name: Name of the runtime to use (default: "thin"). runtime_config: Runtime-specific configuration. status: Current lifecycle status. budget_limit_usd: Optional spending cap in USD. metadata: Arbitrary key-value metadata.

InMemoryAgentRegistry

InMemoryAgentRegistry

In-memory agent registry backed by a plain dict.

Thread-safe for concurrent async access via asyncio.Lock.

register async 

register(record: AgentRecord) -> None

Register an agent. Raises ValueError if id already exists.

get async 

get(agent_id: str) -> AgentRecord | None

Return agent by id, or None if not found.

list_agents async 

list_agents(filters: AgentFilter | None = None) -> list[AgentRecord]

List agents matching optional filters. Returns all if filters is None.

update_status async 

update_status(agent_id: str, status: AgentStatus) -> bool

Update agent status via dataclasses.replace. Returns False if not found.

remove async 

remove(agent_id: str) -> bool

Remove agent from registry. Returns False if not found.