Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Mini-project 7: build a multi-agent system

You have an engine. Now put agents in it.

By the end of this chapter you will have an orchestrator-worker system that decomposes a request into sub-tasks, dispatches them to specialist workers in parallel, collects distilled reports rather than raw transcripts, recovers when a worker’s backend is down, and synthesizes a truthful degraded answer.

The workers draw their tools from three separate MCP servers, each a real subprocess speaking JSON-RPC over stdio. That is where Part 2 comes back, and it is the piece most tutorials skip: a worker is only a specialist if its tool belt is actually different from everyone else’s.

Everything runs offline with a scripted model. Every output block is real.

Setup: we build on the previous chapter’s engine.py, plus three new files:

engine.py          # from mini-project 5, unchanged
mcp_stdio.py       # server helper + client + multi-server hub
orders_server.py   # MCP server 1
kb_server.py       # MCP server 2
policy_server.py   # MCP server 3 — the one that is broken
mas.py             # the multi-agent system

Step 1: three MCP servers

Part 2 built an MCP client. Here we need several servers and one surface over all of them.

A tiny reusable server loop first, in mcp_stdio.py. One JSON-RPC request per line in, one response per line out:

def serve(tools: dict[str, tuple[str, dict, Callable[..., Any]]]) -> None:
    """tools: name -> (description, input_schema, fn)."""
    for line in sys.stdin:
        if not line.strip():
            continue
        req = json.loads(line)
        rid, method, params = req.get("id"), req.get("method"), req.get("params") or {}
        try:
            if method == "tools/list":
                result = {"tools": [{"name": n, "description": d, "inputSchema": s}
                                    for n, (d, s, _) in tools.items()]}
            elif method == "tools/call":
                name = params["name"]
                if name not in tools:
                    raise KeyError(f"unknown tool {name!r}")
                out = tools[name][2](**params.get("arguments", {}))
                result = {"content": [{"type": "text", "text": str(out)}], "isError": False}
            else:
                raise KeyError(f"unknown method {method!r}")
            resp = {"jsonrpc": "2.0", "id": rid, "result": result}
        except Exception as exc:                       # noqa: BLE001
            resp = {"jsonrpc": "2.0", "id": rid,
                    "error": {"code": -32000, "message": f"{type(exc).__name__}: {exc}"}}
        sys.stdout.write(json.dumps(resp) + "\n")
        sys.stdout.flush()

Now three servers, each a few lines. orders_server.py:

def find_order(order_id: str) -> str:
    rec = ORDERS.get(order_id.strip().lstrip("#"))
    return json.dumps(rec) if rec else f"No order {order_id}."

def refund_status(order_id: str) -> str:
    return f"No refund on file for {order_id}; duplicate charge not yet reversed."

serve({
    "find_order": ("Look up an order by ID. Returns customer, item, total, charge count.",
                   _STR, find_order),
    "refund_status": ("Check whether a refund has been issued for an order.",
                      _STR, refund_status),
})

kb_server.py exposes one search_kb tool over a small article table.

policy_server.py exposes check_policy, and it starts fine and then fails every call:

def check_policy(topic: str) -> str:
    raise RuntimeError("policy-db connection refused")

This is deliberate and it is the most useful fixture in the chapter. A backend that is down is easy — you find out at connect time. A backend that is up and broken is the realistic case, and it is the one that exposes whether your system degrades or falls over.

Step 2: a hub over many servers

Each server is its own subprocess and its own pipe. The client is unremarkable:

class MCPClient:
    def __init__(self, name: str, argv: list[str]):
        self.name, self._next_id = name, 0
        self.proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                     stderr=subprocess.DEVNULL, text=True, bufsize=1)

    def request(self, method: str, params: dict | None = None) -> dict:
        self._next_id += 1
        self.proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": self._next_id,
                                          "method": method, "params": params or {}}) + "\n")
        self.proc.stdin.flush()
        line = self.proc.stdout.readline()
        if not line:
            raise ConnectionError(f"server {self.name!r} closed the pipe")
        resp = json.loads(line)
        if "error" in resp:
            raise RuntimeError(f"{self.name}: {resp['error']['message']}")
        return resp["result"]

The interesting part is the hub — one namespaced surface over all the servers:

class ToolHub:
    def __init__(self) -> None:
        self.clients: dict[str, MCPClient] = {}
        self.catalog: dict[str, dict] = {}             # "server.tool" -> spec

    def connect(self, name: str, argv: list[str]) -> None:
        client = MCPClient(name, argv)
        try:
            tools = client.list_tools()
        except Exception as exc:                       # noqa: BLE001
            client.close()
            raise ConnectionError(f"could not connect to {name!r}: {exc}") from exc
        self.clients[name] = client
        for spec in tools:
            self.catalog[f"{name}.{spec['name']}"] = spec

    def specs(self, allow: list[str] | None = None) -> list[dict]:
        out = []
        for qname, spec in self.catalog.items():
            if allow is not None and not any(qname.startswith(p) for p in allow):
                continue
            out.append({"name": qname, "description": spec["description"],
                        "input_schema": spec["inputSchema"]})
        return out

    def call(self, qname: str, arguments: dict) -> str:
        """Never raises: every failure returns text the model can act on."""
        if qname not in self.catalog:
            known = ", ".join(sorted(self.catalog)) or "(none)"
            return f"ERROR: no tool {qname!r}. Available: {known}"
        server, _, tool = qname.partition(".")
        try:
            return self.clients[server].call(tool, arguments)
        except Exception as exc:                       # noqa: BLE001
            return f"ERROR: {qname} failed: {exc}"

Three decisions worth defending.

Namespacing with server.tool. The moment you have more than one MCP server you will have two tools called search. Qualifying the name removes the collision, tells you at a glance where a call went, and — the useful part — makes the allowlist a prefix match.

specs(allow=...) is the allowlist. A worker does not get “all the tools, please only use these.” It gets a filtered list and never sees the others. An unlisted tool is not a rule the model can be talked out of; it is a capability that does not exist from where the model is sitting. This is the cheapest and most durable multi-agent safety control there is.

call never raises. Same rule as Part 1’s registry, now spanning process boundaries. A dead server produces a string the agent can read and reason about, not an exception that kills a fan-out branch.

Connect all three and look at the catalogue:

[hub] 4 tools from 3 servers: ['kb.search_kb', 'orders.find_order',
                               'orders.refund_status', 'policy.check_policy']

Step 3: the handoff contract

Before any agent code, write the boundary. From the previous chapter’s checklist, as two dataclasses:

@dataclass
class Handoff:
    """Everything the worker gets. If it is not in here, the worker cannot see it."""
    sender: str
    recipient: str
    objective: str                       # one sentence, testable
    inputs: dict                         # facts, already resolved
    tool_allow: list[str]                # qualified prefixes: "orders.", "kb."
    max_steps: int = 4
    return_contract: str = "3 bullet findings, each with the tool result it came from"


@dataclass
class Report:
    worker: str
    objective: str
    status: str                          # "ok" | "failed"
    findings: list[str] = field(default_factory=list)
    note: str = ""
    steps_used: int = 0

Read the docstring on Handoff again, because it is the design. The worker’s entire world is this object plus its allowed tools. It cannot reach into the supervisor’s state, cannot see the original user message, cannot see what its sibling workers found. That is context isolation enforced by construction rather than by hoping.

Report is the return contract as a type. status in particular: the supervisor must be able to tell success from failure without reading prose, because a supervisor that parses “I wasn’t able to…” out of a paragraph will eventually parse it wrong.

Step 4: the worker agent

A worker is the Part 1 loop with three additions: an allowlist, a budget that came from the packet, and a structured return.

class Worker:
    def __init__(self, name, model, hub, verbose=True):
        self.name, self.model, self.hub, self.verbose = name, model, hub, verbose

    def run(self, packet: Handoff) -> Report:
        visible = [s["name"] for s in self.hub.specs(allow=packet.tool_allow)]
        self.log(f"  <{self.name}> objective: {packet.objective}")
        self.log(f"  <{self.name}> tools: {visible}")

        observations: list[str] = []
        for step in range(1, packet.max_steps + 1):
            turn = self.model.next_turn(self.name, observations)
            if turn.final is not None:
                return Report(self.name, packet.objective, "ok",
                              findings=observations, note=turn.final,
                              steps_used=step - 1)
            for call in turn.calls:
                if not any(call.tool.startswith(p) for p in packet.tool_allow):
                    obs = f"ERROR: {call.tool} is not in this agent's allowlist"
                else:
                    obs = self.hub.call(call.tool, call.args)
                self.log(f"  <{self.name}> [{step}] {call.tool}"
                         f"({json.dumps(call.args)}) -> {obs[:70]}")
                observations.append(f"{call.tool}: {obs}")
        return Report(self.name, packet.objective, "failed", findings=observations,
                      note=f"exhausted {packet.max_steps}-step budget without an answer",
                      steps_used=packet.max_steps)

Notice the allowlist is checked twice, in two different places, on purpose. specs(allow=...) decides what the model is told about. The startswith check decides what actually executes. Filtering the advertised list is a hint; the execution check is the control. Models call tools they were not offered — you saw it in Part 1 — so never rely on the list alone.

Notice too what the worker returns on failure: status="failed" and a normal Report. It does not raise. A worker exhausting its budget is a business outcome the supervisor must handle, exactly like a step cap firing in Part 1.

The model here is scripted, one script per worker:

class ScriptedModel:
    def __init__(self, scripts: dict[str, list[Turn]]):
        self.scripts = {k: list(v) for k, v in scripts.items()}

    def next_turn(self, worker: str, observations: list[str]) -> Turn:
        script = self.scripts.get(worker) or []
        if not script:
            return Turn(final="(no plan left)")
        return script.pop(0)

Swap this for a real client and nothing else in the file changes. It is a mock because determinism is what lets us assert on the output — and because a multi-agent system that only behaves correctly when the model is having a good day is one you cannot test.

Step 5: the supervisor as a graph

Now the part the engine makes easy.

@dataclass
class Mission:
    request: str
    plan: list[Handoff] = field(default_factory=list)
    reports: list[Report] = reduced(list, operator.add)
    answer: str = ""

One reduced field: reports. Three workers finish in the same superstep and all three reports survive, in dispatch order. Without that reducer, two of them vanish — and this is the bug you would have shipped if you had written the fan-out yourself with a dict.

The nodes:

    @wf.node("plan")
    def plan(s: Mission) -> dict:
        packets = [
            Handoff("supervisor", "billing",
                    "Establish what the customer was actually charged and whether "
                    "a refund exists.",
                    {"order_id": "12345"}, ["orders."], max_steps=3),
            Handoff("supervisor", "policy",
                    "State the refund policy that applies to a duplicate charge.",
                    {"topic": "duplicate charge"}, ["policy."], max_steps=2),
            Handoff("supervisor", "comms",
                    "Find the customer-facing wording for duplicate charges.",
                    {"query": "duplicate charge"}, ["kb."], max_steps=2),
        ]
        return {"plan": packets}

    @wf.node("dispatch")
    def dispatch(s: Mission, packet: Handoff) -> dict:
        return {"reports": [workers[packet.recipient].run(packet)]}

    @wf.node("recover")
    def recover(s: Mission) -> dict:
        failed = [r for r in s.reports if r.status == "failed"]
        out = []
        for r in failed:
            out.append(Report("supervisor", r.objective, "ok",
                              findings=["fallback: cached policy snapshot 2026-07"],
                              note="Refunds for duplicate charges: automatic within "
                                   "5 business days, goodwill credit beyond that.",
                              steps_used=0))
        return {"reports": out}

    @wf.node("synthesize")
    def synthesize(s: Mission) -> dict:
        good = [r for r in s.reports if r.status == "ok"]
        degraded = any(r.worker == "supervisor" for r in good)
        body = " | ".join(f"{r.worker}: {r.note}" for r in good)
        flag = ("  [DEGRADED: one specialist failed; used a cached fallback]"
                if degraded else "")
        return {"answer": f"{body}{flag}"}

plan here builds the packets in code. In a live system this node is a model call with a structured output schema producing exactly this list — and the reason to keep it in one node with a typed output is that you can then evaluate decomposition quality on its own, separately from whether the workers did their jobs.

Each packet’s budget and allowlist are set by the supervisor, per sub-task. billing gets three steps and the orders server; policy gets two steps and only the policy server. Budgets belong to the task, not to the agent.

The wiring is four lines:

    def fan_out(s: Mission):
        return [Send("dispatch", p) for p in s.plan]

    def needs_recovery(s: Mission):
        return "recover" if any(r.status == "failed" for r in s.reports) else "synthesize"

    (wf.start("plan")
       .branch("plan", fan_out)
       .edge("dispatch", "collect")
       .branch("collect", needs_recovery)
       .edge("recover", "synthesize")
       .edge("synthesize", END))

That is the entire orchestrator-worker topology: one fan-out over a runtime-sized plan, one join, one conditional recovery edge, one synthesis.

The needs_recovery router is the important line. Recovery is a structural decision made by deterministic code reading a typed status field. It is not a paragraph in the supervisor’s prompt asking it to please notice when a worker fails. Prompts are advisory; edges are not.

Run it

$ python3 mas.py

[hub] 4 tools from 3 servers: ['kb.search_kb', 'orders.find_order', 'orders.refund_status', 'policy.check_policy']

[superstep 1] plan
[plan] 3 sub-tasks: ['billing', 'policy', 'comms']
        plan             -> ['plan']
[superstep 2] dispatch, dispatch, dispatch
  <billing> objective: Establish what the customer was actually charged and whether a refund exists.
  <billing> tools: ['orders.find_order', 'orders.refund_status']
  <billing> [1] orders.find_order({"order_id": "12345"}) -> {"customer": "R. Okafor", "item": "Solaris ANC headphones", "total": "
  <policy> objective: State the refund policy that applies to a duplicate charge.
  <policy> tools: ['policy.check_policy']  <billing> [1] orders.refund_status({"order_id": "12345"}) -> No refund on file for 12345; duplicate charge not yet reversed.
  <comms> objective: Find the customer-facing wording for duplicate charges.
  <comms> tools: ['kb.search_kb']
  <comms> [1] kb.search_kb({"query": "duplicate charge"}) -> KB-201: Duplicate authorisations clear in 5 business days; issue a goo

  <policy> [1] policy.check_policy({"topic": "duplicate charge"}) -> ERROR: policy.check_policy failed: policy: RuntimeError: policy-db con
  <policy> [2] policy.check_policy({"topic": "duplicate charge"}) -> ERROR: policy.check_policy failed: policy: RuntimeError: policy-db con
        dispatch         -> ['reports']
        dispatch         -> ['reports']
        dispatch         -> ['reports']
[superstep 3] collect
[collect] billing  ok     2 finding(s), 1 step(s)
[collect] policy   failed 2 finding(s), 2 step(s)
[collect] comms    ok     1 finding(s), 1 step(s)
        collect          -> (no update)
[superstep 4] recover
[recover] policy failed: exhausted 2-step budget without an answer
        recover          -> ['reports']
[superstep 5] synthesize
        synthesize       -> ['answer']

ANSWER: billing: Two charges of $129.00 on order 12345; no refund issued yet. | comms: Tell the customer the duplicate clears in 5 business days. | supervisor: Refunds for duplicate charges: automatic within 5 business days, goodwill credit beyond that.  [DEGRADED: one specialist failed; used a cached fallback]
wall clock: 0.02s

Six things in that output are worth stopping on.

The tool belts are genuinely different. billing sees two tools from the orders server, policy sees one from the policy server, comms sees one from the kb server. Nobody sees all four. Three MCP subprocesses, one namespaced hub, three disjoint views.

The fan-out is real. Superstep 2 dispatched three workers concurrently.

The interleaved log lines are a lesson, not a bug. Look at <policy> tools: [...] and <billing> [1] orders.refund_status(...) colliding on one line. That is what print does from three threads. It is exactly why production systems emit structured events with a run ID and a span ID rather than lines to stdout — the first thing you lose to concurrency is a readable trace. Fix this before you fix anything else in a real build.

The failure was contained. policy burned both its steps on a backend returning RuntimeError: policy-db connection refused, then returned status="failed". It did not raise, it did not retry forever, and it did not stop billing or comms.

The supervisor recovered structurally. collect saw one failed status, the conditional edge routed to recover, and a cached fallback entered reports as a normal report.

The answer is honest about it. [DEGRADED: one specialist failed; used a cached fallback]. This is the behaviour you want and almost never get for free: a system that produces a useful answer under partial failure and says so. The alternative — silently substituting a cached policy and presenting it as current — is worse than an error, because nobody downstream can tell.

What this system still gets wrong

The honest inventory, same as Part 1.

The supervisor’s plan is hardcoded. In reality plan is a model call, which means decomposition can be wrong: overlapping sub-tasks, a missing one, or one whose objective is too vague to be testable. That is the highest-severity failure in the whole architecture and it needs its own eval set.

No shared-fact resolution. If two workers both needed the order record they would both fetch it. Resolve shared facts in plan and put them in every packet’s inputs.

No cost budget. max_steps per worker caps calls, not money. A worker doing three steps over huge documents costs more than another doing six over small ones. Add a token counter to Report and check the total at the superstep barrier.

Findings are not verified. The supervisor takes every note at face value. Workers do carry their raw tool observations in findings, so the raw material for a verification node exists — but nothing checks that a note is supported by them.

Recovery is one strategy. A cached fallback is one option. Reassigning to a different worker, retrying with a longer budget, or escalating to a human are others, and which one applies depends on why the worker failed. Report should carry a failure reason code so needs_recovery can route on it.

No per-worker timeout. A worker that hangs hangs its superstep. The engine’s wall-clock budget catches it eventually, but only at the next barrier.

MCP servers are launched per run and never health-checked. Real deployments keep pooled connections, health-check on an interval, and mark a server degraded so the supervisor can plan around it rather than discovering it mid-fan-out.

Handoffs are one-way. A worker cannot ask a clarifying question. When the objective is ambiguous it guesses. Adding a needs_clarification status and a supervisor edge that handles it is a genuinely useful exercise.

Extensions worth doing

Reason codes and routed recovery. Add failure: Literal["budget", "tool_error", "ambiguous", "refused"] to Report and make needs_recovery a real router with a branch per code.

A verification node. Between collect and synthesize, add a node that checks each note against the worker’s own findings and flags unsupported claims. This is the structural fix for cascading errors.

Real MCP servers. Swap the fixtures for the official Python SDK (https://github.com/modelcontextprotocol/python-sdk) and point one worker at a third-party server. The hub does not change; that is the point of the protocol.

The same system in LangGraph. plan returns [Send("dispatch", packet) for packet in packets], reports becomes Annotated[list[Report], operator.add], and needs_recovery becomes an add_conditional_edges router. Compile with a checkpointer and the recovery path becomes resumable. LangGraph also ships a prebuilt supervisor (https://docs.langchain.com/oss/python/langchain/multi-agent) — read its source and compare it with what you built.

Peer handoff for comparison. Rebuild the routing half in the OpenAI Agents SDK, where handoffs are tools the model calls (https://openai.github.io/openai-agents-python/handoffs/). Feeling the difference between “supervisor collects results” and “control moves and does not come back” is worth an hour.

Tracing. Emit one structured JSON event per node and per worker step, with run ID, superstep, worker, packet hash, and outcome. Then re-read the interleaved output above and appreciate what you have fixed.

What you should be able to do now

  • Build an orchestrator-worker system on a graph engine: plan, runtime-sized parallel fan-out, join, conditional recovery, synthesis.
  • Define a typed handoff packet and a typed report, and explain what each field prevents — objective ambiguity, unresolved references, unbounded workers, prose the supervisor has to parse.
  • Aggregate tools from several MCP servers behind one namespaced hub, and give each worker a disjoint tool belt enforced both at advertisement and at execution.
  • Make a worker fail safely: budget exhausted, structured failure status, no exception across the boundary, siblings unaffected.
  • Route recovery on a typed status field with a conditional edge rather than asking a supervisor prompt to notice failures.
  • Produce a degraded answer that is explicit about being degraded, and say why that is the required behaviour rather than a nicety.
  • Name what your system still gets wrong — plan quality, unverified findings, cost budgets, one-way handoffs — and describe the concrete fix for each.

Further reading