Build a Multi-Agent Workflow in Python with Clear Handoffs

Breaking one task into several agents does not automatically make it reliable. The agents also need clear responsibilities, a safe way to exchange data, and one place that decides what runs next.

This tutorial builds those parts before adding a language model. Three small Python agents process fictional equipment reimbursement claims. An intake agent validates the amount and normalizes each claim. A policy agent checks simple rules. A decision agent converts the findings into one outcome. A supervisor controls their order and stops early when the amount is invalid.

Every result is deterministic: the same claim always follows the same route. This lets us inspect multi-agent orchestration without an API key, model cost, or changing output. Later, a model could replace one specialist while the handoff contracts and supervisor remain testable Python.

This lesson is about coordination between specialists. If you first need to control repeated tool calls from one model, read Build a Safe AI Agent Loop in Python.

Prerequisites and Setup

You need Python 3.10 or later and basic knowledge of functions, dictionaries, and if statements. The workflow itself uses only Python’s standard library. Matplotlib is needed to rebuild the outcome chart.

Create a virtual environment and install Matplotlib:

python -m venv .venv
source .venv/bin/activate
python -m pip install matplotlib

On Windows PowerShell, use .venv\Scripts\Activate.ps1 for the activation command.

The executed build used Python 3.13.2 and Matplotlib 3.11.0. You can download the eight fictional reimbursement claims. This original synthetic dataset is released under CC0 1.0. It contains no real people, employers, purchases, or policy decisions.

Mental Model: Specialists Connected by Contracts

An agent is a component that receives a goal and context, performs a bounded job, and produces a result for the next part of a system. An agent often uses a large language model, but a model is not required to learn or implement the orchestration pattern.

In this lesson, each agent is represented by a narrow Python function. Each function can read or update the workflow state, but it has one responsibility:

raw claim
   |
   v
intake agent -- invalid input --> stop
   |
   v
policy agent
   |
   v
decision agent --> final status

The arrow between two agents represents a handoff, a defined boundary where one specialist’s output becomes another specialist’s input. In this code-controlled workflow, the supervisor retains control and passes the shared state to each function. The typed NormalizedClaim inside that state is the data contract: it defines which normalized fields later agents may expect.

The supervisor uses code orchestration, which means ordinary Python selects the next step. Current OpenAI Agents SDK documentation distinguishes this from model-led orchestration and notes that code orchestration is useful when predictable speed, cost, and behavior matter. It also describes two model-led patterns: a manager that keeps control while calling specialists, and handoffs where a specialist takes over. See the official agent orchestration guide for those current patterns.

Our code-controlled chain is a good starting point when the business route is already known. It would be unnecessary to ask a model whether validation should happen before a policy check.

Create a Typed Handoff Contract

Raw CSV values are strings. Later agents should not repeatedly guess whether "92.00" is a number or whether "yes" means a receipt exists. The intake agent will convert those strings into one typed object.

A Python dataclass is a class designed mainly to store data. frozen=True prevents its fields from being reassigned after creation, which keeps the normalized handoff stable:

from dataclasses import dataclass, field

@dataclass(frozen=True)
class NormalizedClaim:
    claim_id: str
    item: str
    amount: float
    has_receipt: bool
    category: str

The workflow also needs mutable state. It holds the raw input, the optional normalized claim, policy findings, final status, and a trace. A trace is an ordered record of what ran and what happened.

@dataclass
class WorkflowState:
    raw: dict[str, str]
    normalized: NormalizedClaim | None = None
    policy_findings: list[str] = field(default_factory=list)
    status: str = "started"
    trace: list[dict[str, str]] = field(default_factory=list)

default_factory=list creates a fresh list for every workflow run. Writing policy_findings=[] as a class default would let multiple instances share one mutable list, which can mix results between claims.

The two classes have different purposes. NormalizedClaim is the stable message produced by intake. WorkflowState is the changing record owned by the supervisor. This separation makes the handoff visible and prevents every agent from inventing a different data shape.

Build Three Specialist Agents

The intake agent is the only component that reads raw amount text. It catches a missing or non-numeric amount, records a clear stop event, and returns without raising a deep error.

def intake_agent(state: WorkflowState) -> None:
    try:
        amount = float(state.raw["amount"])
    except (KeyError, ValueError):
        state.status = "invalid_input"
        state.trace.append({
            "agent": "intake",
            "event": "stopped",
            "detail": "amount is not numeric",
        })
        return

    if amount <= 0:
        state.status = "invalid_input"
        state.trace.append({
            "agent": "intake",
            "event": "stopped",
            "detail": "amount must be positive",
        })
        return

    state.normalized = NormalizedClaim(
        claim_id=state.raw["claim_id"],
        item=state.raw["item"].strip(),
        amount=amount,
        has_receipt=state.raw["receipt"].lower() == "yes",
        category=state.raw["category"].lower(),
    )
    state.trace.append({
        "agent": "intake",
        "event": "completed",
        "detail": "normalized claim",
    })

After this function succeeds, later agents read amount as a float and has_receipt as a Boolean. They no longer need to know how those values were represented in the CSV file.

The policy agent can now focus on one fictional lesson policy. Equipment and training are covered categories. A receipt is required. Amounts above 500 need a person to review them.

def policy_agent(state: WorkflowState) -> None:
    claim = state.normalized
    assert claim is not None

    if claim.category not in {"equipment", "training"}:
        state.policy_findings.append("category is not covered")
    if not claim.has_receipt:
        state.policy_findings.append("receipt is missing")
    if claim.amount > 500:
        state.policy_findings.append("amount needs manager review")

    state.trace.append({
        "agent": "policy",
        "event": "completed",
        "detail": "; ".join(state.policy_findings) or "no findings",
    })

The assertion checks a handoff invariant: the policy agent requires a normalized claim. The supervisor must never call it after failed intake. In a public API, an explicit exception with a descriptive message may be more suitable, but this assertion is useful in our controlled build and tests.

The decision agent translates findings into one status. It does not parse raw data or repeat policy thresholds.

def decision_agent(state: WorkflowState) -> None:
    if (
        "category is not covered" in state.policy_findings
        or "receipt is missing" in state.policy_findings
    ):
        state.status = "needs_correction"
    elif "amount needs manager review" in state.policy_findings:
        state.status = "manual_review"
    else:
        state.status = "auto_approved"

    state.trace.append({
        "agent": "decision",
        "event": "completed",
        "detail": state.status,
    })

The rule order is intentional. A missing receipt needs correction even when the amount is also high. The decision agent gives that correctable problem priority over manual review. Real policy rules should document precedence in the same way.

Let a Supervisor Own the Route

The supervisor is short because the specialists hold the detailed work. Its important job is to enforce the order and the early stop:

def run_workflow(raw_claim: dict[str, str]) -> WorkflowState:
    state = WorkflowState(raw=raw_claim)

    intake_agent(state)
    if state.status == "invalid_input":
        return state

    policy_agent(state)
    decision_agent(state)
    return state

This function is the only entry point the batch runner needs. It guarantees that the policy agent is not called without a NormalizedClaim. It also returns the complete state, not only the final status, so a developer can inspect why the workflow stopped.

Run it first with one small claim:

example = {
    "claim_id": "C03",
    "item": "Desk lamp",
    "amount": "71.25",
    "receipt": "no",
    "category": "equipment",
}

result = run_workflow(example)
print("status:", result.status)
for event in result.trace:
    print(event["agent"], "->", event["event"], "->", event["detail"])

The executed result was:

status: needs_correction
intake -> completed -> normalized claim
policy -> completed -> receipt is missing
decision -> completed -> needs_correction

Read this from top to bottom. Intake successfully converted the raw strings. Policy found one specific problem. Decision mapped that finding to needs_correction. The trace explains the route without exposing an internal model prompt or relying on a vague final sentence.

Execute and Verify Every Route

One successful case does not show whether the stop rule works. The dataset contains eight claims that exercise four outcomes. The build runs the same supervisor for every row and compares each result with an expected status.

expected = {
    "C01": "auto_approved",
    "C02": "auto_approved",
    "C03": "needs_correction",
    "C04": "auto_approved",
    "C05": "manual_review",
    "C06": "invalid_input",
    "C07": "auto_approved",
    "C08": "needs_correction",
}

runs = [run_workflow(claim) for claim in claims]

for run in runs:
    assert run.status == expected[run.raw["claim_id"]]
    expected_steps = 1 if run.status == "invalid_input" else 3
    assert len(run.trace) == expected_steps

The second assertion tests control flow, not only final statuses. Invalid claim C06 must stop after intake. If the supervisor called policy or decision after that failure, the route-length assertion would fail even if later code produced the expected status by accident.

Here is the real console output from the complete build:

cases=8 auto_approved=4 needs_correction=2 manual_review=1 invalid_input=1
C01: auto_approved    route=intake -> policy -> decision
C02: auto_approved    route=intake -> policy -> decision
C03: needs_correction route=intake -> policy -> decision
C04: auto_approved    route=intake -> policy -> decision
C05: manual_review    route=intake -> policy -> decision
C06: invalid_input    route=intake
C07: auto_approved    route=intake -> policy -> decision
C08: needs_correction route=intake -> policy -> decision

C06 is the key control-flow test. Its amount is not-recorded, so only intake appears in its route. The other seven claims reach all three agents.

The chart summarizes outcomes, while the trace preserves individual routes. Keep both views separate. Aggregate counts help monitor a batch; per-run traces help explain and debug one result.

When to Add a Language Model

This workflow does not need a model because its rules are precise. Adding one to numeric conversion or a fixed threshold would make a known operation less predictable.

A model may help when an agent receives unstructured text. For example, an intake specialist could extract item, amount, and category from an email. Keep its output behind the same NormalizedClaim contract and validate every field before policy runs. The model proposes structured data; Python decides whether the proposal is valid.

Another design choice is who owns the final response. In a manager pattern, one agent calls specialists and combines their results. In the Agents SDK’s handoff pattern, a specialist takes over the conversation. That transfer of control differs from this tutorial’s code-controlled data handoffs, where the Python supervisor remains in charge. The current OpenAI Agents SDK handoff documentation explains that framework handoffs can use typed input and local validation, while the agents guide compares manager and handoff designs.

Whichever framework you choose, retain three properties from this lesson:

1. Give each specialist a narrow job.
2. Validate a typed boundary between specialists.
3. Keep routing, stop conditions, and traces inspectable.

Common Mistakes and Troubleshooting

Every function is called an agent. A useful specialist needs a clear responsibility and contract. If several functions always change together and cannot be tested separately, they may be one component split into too many names.

Agents pass free-form text to one another. Define named fields for required data. Validate model-generated or external values at the boundary. Free-form summaries are useful for people, but fragile as the only machine-to-machine contract.

The supervisor calls policy after failed intake. Check the stop status immediately after intake. Add a route-length assertion so tests prove that downstream agents did not run.

Rules are duplicated across agents. Keep conversion in intake, eligibility checks in policy, and outcome mapping in decision. Duplicated thresholds can drift and produce conflicting results.

The trace contains sensitive data. Our fictional trace stores only agent names, events, and short findings. A production trace may contain personal or financial information. Minimize recorded fields, restrict access, redact secrets, and set a retention period.

Parallel execution is added to a dependent chain. Policy needs intake output, and decision needs policy findings. These steps cannot safely run at the same time. Use parallel execution only for specialists whose inputs do not depend on each other’s results.

assert is used as user-input validation. In this tutorial, assert claim is not None checks a programmer-controlled invariant after the supervisor has validated the route. Do not use assertions to validate external data because Python can disable them. Return a status or raise a normal exception for user input.

Recap

A multi-agent workflow is more than a list of specialists. It needs an explicit route and reliable handoffs.

We created a frozen NormalizedClaim as the handoff contract and a mutable WorkflowState as the run record. Intake converted raw strings, policy produced focused findings, and decision chose one status. The supervisor enforced dependency order and stopped invalid input before later agents ran. Finally, eight deterministic cases verified both outcomes and route lengths.

Start with this code-controlled structure when the process is known. If you later add a language model for an uncertain step, keep the typed boundary, validation, stop rules, and trace in ordinary Python. Those parts make specialist behavior easier to test, replace, and understand.