multi-agent-orchestrator

A small, dependency-free template for routing work to capability-based agents. The pattern behind most agentic systems, distilled to its essentials:

• Agent - a worker that advertises one or more capabilities and handles tasks.
• AgentRegistry - the single source of truth: register agents, discover them by capability, resolve aliases to canonical names, probe health.
• Orchestrator - routes a task to a capable agent (round-robin when several qualify), runs it with a timeout, wraps the outcome in a TaskResult, and emits a lifecycle event. fan_out runs every capable agent concurrently.
• AsyncMessageBus - an in-memory pub/sub so the orchestrator can broadcast task.completed / task.failed without coupling to a broker. Swap it for MQTT, NATS, or Redis Streams in production.

Pure standard library + asyncio. No external dependencies.

Generalized from the orchestration layer of Talos, my agentic platform, where a control plane routes work across dozens of specialized agents.

Usage

import asyncio
from orchestrator import Agent, AgentRegistry, Orchestrator, Task

class UpperAgent(Agent):
    def __init__(self):
        super().__init__("upper", capabilities={"text.transform"})
    async def handle(self, task):
        return task.payload["text"].upper()

async def main():
    registry = AgentRegistry()
    registry.register(UpperAgent(), aliases=["uppercase"])
    orch = Orchestrator(registry)

    result = await orch.submit(Task(id="t1", capability="text.transform", payload={"text": "hello"}))
    print(result.agent, result.output)   # upper HELLO

asyncio.run(main())

Routing is by capability, not by hard-coded wiring: register a second agent for text.transform and the orchestrator round-robins between them; ask for a capability nobody serves and you get a clean failed TaskResult instead of an exception. Agents that raise or exceed the timeout also come back as failed results, never crashing the caller.

Demo & tests

python -m examples.demo   # routing, round-robin, fan-out, bus events
pytest                    # registry + orchestrator behavior

License

MIT