Skip to content

Multi-agent workflows

Multi-agent workflows are first-class in Tulip: eight shapes you compose in one process or scale across a mesh, every shape backed by the same Agent class, the same event stream, and the same primitives. Compose the shape that fits the work.

Multi-agent shapes Nine panels, one per shape. Sequential: Research, Write and Edit run in order, each output prompting the next, and the last output is the result. Parallel: one task goes to three agents at once and their outputs are joined into one result. Loop: an Editor runs a pass and, while its condition is not met, is re-prompted with its own output; once it is met, or at max_loops, that output is the result. Orchestrator and Specialists: the orchestrator routes subtasks to the specialists it picks (it can leave any of them out, as the dashed one shows), then correlates their findings into a summary. Swarm: with no router, agents claim tasks from a priority task queue when their tags match, and read and write a SharedContext of findings and a blackboard. Handoff: a Triage agent passes a HandoffContext holding the task, a conversation summary and findings, not the full transcript, to a Billing agent, and the number of hand-offs is capped. StateGraph: Draft updates shared state, then Review's router sends the work back to Draft to revise or on to END once approved; loops stop at max_iterations. Functional API: an @entrypoint function awaits @task functions, which can add retries and caching. A2A: an A2AClient in one process sends JSON-RPC to an A2AServer in another and gets a result back, or an SSE stream when it asks for a streamed reply; the server publishes an AgentCard and runs the Agent. COMPOSITION · SEQUENTIAL COMPOSITION · PARALLEL COMPOSITION · LOOP ORCHESTRATOR + SPECIALISTS SWARM HANDOFF STATEGRAPH FUNCTIONAL API A2A output output same task not met met, or max_loops subtask claim approved revise await JSON-RPC result or SSE Research Write Edit Result Task Web Data Docs Result Editorone pass Result Orchestratorroutes subtasks Logs Metrics Traces Summarycorrelated findings Task queueby priority Scout Verify Write SharedContextfindings · blackboard TriageL1 support HandoffContexttask · summary · findings BillingL2 specialist Draftupdates state Reviewrouter END workflow@entrypoint fetch@task draft@task check@task A2AClientprocess A A2AServerprocess B Agent Each output, plus the task, prompts the next agent. The last output is the result. Every agent gets the same task at once; the outputs are joined into one result. Re-prompts the agent with its own output until the condition holds, or max_loops. Its model routes subtasks to the specialists it picks, then correlates and summarizes. No router: each agent claims queued tasks its tags match, and shares what it finds. Carries a summary, not the full transcript; the number of hand-offs is capped. Nodes update shared state; a router picks the next node. Loops stop at max_iterations. Plain async functions: @task can add retries and caching; fan out with asyncio.gather. A2AServer serves an AgentCard; clients in any process or runtime call it over HTTP.

Don't know which shape to use?

See Pick a shape — three questions get you there. For a straight chain or fan-out, start with Composition.

What you can ship today

Every example below is a real examples/notebook_NN_*.py file in the repo, runs end-to-end against the bundled MockModel (no creds), and upgrades to a live provider by setting one env var.

Workflow One line Code
29 DeepAgent — research factory create_deepagent with reflexion + grounding + subagent dispatch + deepagent.* SSE events. notebook_29_deepagent.py
30 Customer-support triage at scale Scatter 3 tickets × 3 lenses (sentiment, routing, resolution) to 9 analyst agents via Send, reduce into one triage summary. notebook_30_map_reduce_code_review.py
31 Supervisor + critic loop Recon → Report author → Skeptical reviewer, loop back to the author until the reviewer approves (cap'd revisions). notebook_31_supervisor_critic_loop.py
32 Adversarial debate + judge One agent argues the finding is a true positive, another argues benign, across N rounds; Judge emits a typed Verdict via output_schema. notebook_32_debate_with_judge.py
33 Multi-agent + human-in-the-loop Three patterns in one file: approval gate, human-as-tool, long-pause snapshot/resume. notebook_33_multiagent_human_in_loop.py
63 IR (incident-response) war-room Triage → 3 parallel investigators (SIEM — a security team's log platform — EDR, and threat intel) → severity gate → page-the-responder → contain → typed Postmortem. notebook_63_incident_response.py
64 Customer-support concession approval Ticket analyst → Impact analyst → risk-tier router (auto / support manager / +billing / +director) → typed ConcessionDecision. Stacked interrupt() gates on the top tiers. notebook_64_procurement_approval.py
65 DPA & security-addendum review Parser → 3 parallel reviewers (privacy / security / compliance) → revision gate → human analyst → Command(goto="sign_off") short-circuits when resolved. Cycles enabled. notebook_65_contract_review.py

Pick a shape

Three questions get you to the right shape almost every time:

  1. Do agents need to talk across processes or runtimes? If yes, you want A2A. If no, everything else lives in one Python process.
  2. Does the flow have cycles or conditional routing? If yes, you want StateGraph. If it's a straight chain or fan-out, you want Composition.
  3. Do you want one coordinator picking the next agent, or peers collaborating without a central router? Coordinator → Orchestrator + Specialists. Peers → Swarm. A single agent passes the conversation onward → Handoff.

The decision tree below is the same questions in diagram form.

Pick a multi-agent shape Three questions pick the shape. First, do agents need to talk across processes or runtimes? If yes, use A2A. If no, everything runs in one process, and the second question is the shape of the flow: a straight chain or fan-out is Composition, cycles or branches are StateGraph, and if the agents decide as they go, the third question is who picks the next agent. One coordinator is Orchestrator plus Specialists, peers claiming tasks is Swarm, and the current agent passing its findings on is Handoff. yes no chain or fan-out cycles or branches agents decide one coordinator peers claim tasks the current agent Talk across processes? or across runtimes What shape is the flow? all in one process Who picks the next agent? no fixed flow A2A remote agents over HTTP Composition pipelines of agents StateGraph nodes, edges, state Orchestrator + Specialists routes and summarizes Swarm shared context Handoff passes its findings on

Writing your own glue (asyncio fan-out, retries, schedulers)? Use the Functional API (@task, @entrypoint) — a thin wrapper that brings agent runs into the ordinary asyncio universe.

The eight shapes

Eight shapes across seven pattern pages — Composition covers three of them (Sequential, Parallel, Loop) in one page, and the Functional API rides alongside as the asyncio-native escape hatch.

Pattern Best for Key class Source
Composition linear chains; fan-out + merge; revise-until-confidence SequentialPipeline, ParallelPipeline, LoopAgent agent/composition.py
Orchestrator + Specialists one router decides which expert handles each sub-task Orchestrator, Specialist multiagent/orchestrator.py
Swarm open-ended research; peer-to-peer; shared context Swarm, SharedContext multiagent/swarm.py
Handoff escalation desks; investigation moves with a findings + progress summary Handoff, HandoffAgent, create_handoff_manager multiagent/handoff.py
StateGraph explicit DAG with cycles, conditional edges, subgraphs StateGraph, Node, Edge multiagent/graph.py
Functional map/reduce over agents; asyncio-native composition @task, @entrypoint multiagent/functional.py
A2A cross-process / cross-runtime; capability discovery A2AServer, A2AClient, AgentCard a2a/protocol.py

Workflow primitives

The pieces every shape is built from. Drop them into any graph node.

Send — scatter / map-reduce

from tulip.core.send import Send
async def split(state):
    return [Send("worker", {"task": t}) for t in state["tasks"]]

Returning a list of Send from a node spawns parallel executions — no asyncio.gather, no shared mutable state. Each result lands in state[send.send_id] keyed by the send id.

interrupt() — pause for a human

from tulip.core import interrupt

async def approval_node(state):
    response = interrupt({"question": "Ship it?", "options": ["yes", "no"]})
    return {"approved": response == "yes"}

interrupt() raises InterruptException; the graph catches it, snapshots state, and returns control to the caller. Resume by calling graph.execute(Command(resume="yes")).

Command(goto=...) — explicit routing

from tulip.core import goto

async def smart_router(state):
    if state["urgent"]:
        return goto("emergency", priority=10)   # skip ahead
    return {"score": compute_score(state)}      # normal flow

Return a Command from a node to override the default edge — useful for short-circuiting refinement loops or skipping straight to sign-off. Used to skip the revision loop when the analyst says RESOLVED.

Agent(output_schema=...) — typed terminal artifacts

from pydantic import BaseModel
from tulip.agent import Agent, AgentConfig
class Verdict(BaseModel):
    disposition: str   # "true_positive" | "benign"
    confidence: float
    reasoning: str

agent = Agent(config=AgentConfig(model="openai:gpt-4o-mini", output_schema=Verdict))
result = agent.run_sync("...")
verdict: Verdict = result.parsed   # validated Pydantic instance, not free text

When you need a typed artifact at the workflow boundary — Verdict, Postmortem, ConcessionDecision, ContractDecision — output_schema gives you a validated Pydantic instance.

GraphConfig(allow_cycles=True) — refinement loops

from tulip.multiagent.graph import GraphConfig, StateGraph
graph = StateGraph(config=GraphConfig(allow_cycles=True, max_iterations=20))
graph.add_edge("critic", "writer")   # loop edge — only legal with allow_cycles

Cycles are off by default (so you can't accidentally infinite-loop). Opt in with allow_cycles=True plus an iteration cap.

Why these workflows ship to prod

The boring stuff that turns a demo into a product — typed terminal artifacts, idempotent tools, checkpointing, reflexion, grounding, and streaming — works in any of the eight shapes; you don't pick "shape" or "production-ready", you get both. And a side-effecting tool in any shape can be put behind the admission gate — wrap it with gate_tool(...) or call admit() in its body, passing your ControlPolicy and an AuditTrail. Every decision on it, whether allow, hold or deny, is then appended to that hash-chained trail: each entry commits to the one before it, so an edit, reorder, or deletion in the middle breaks verification. Sign the trail and verify it against the public key (verify(keys=...)), and a chain rebuilt around an edit by anyone without your signing key fails too.

→ Production-readiness

One event stream across all of them

All eight shapes — A2A included — share the same typed event taxonomy. Consume directly from the generator, or use the opt-in EventBus to get per-component SSE events (agent.think, agent.tool.started, multiagent.orchestrator.routing, etc.) from every layer simultaneously:

from tulip.observability import run_context, get_event_bus

async with run_context() as rid:
    result = await orchestrator.execute("Resolve the disputed orders in this morning's queue.")
    # The run is done: close its stream so the subscriber below replays
    # the recorded history and then stops, instead of waiting for more.
    await get_event_bus().close_stream(rid)

    async for ev in get_event_bus().subscribe(rid):
        match ev.event_type:
            case "multiagent.orchestrator.decision" if "specialists_selected" in ev.data:
                print("orchestrator →", ev.data["specialists_selected"])
            case "multiagent.specialist.completed":
                print("  ✓", ev.data["specialist_type"], ev.data.get("output_preview", ev.data.get("error")))

Orchestrator events carry orchestrator_id and specialist events carry specialist_id and specialist_type, so you can attribute output to the specialist that produced it. SSE streams from AgentServer carry the same shape — your front-end consumer is unchanged whether the back-end is a single agent, an orchestrator, a swarm, or an A2A mesh.

→ Observability — EventBus & SSE · SSE event catalogue

Mixing shapes

A StateGraph node runs any async callable, so one node can await orchestrator.execute(...), another swarm.execute(...), and a third can be a whole StateGraph added as a subgraph. Give a node a retry_policy and the graph re-runs it with backoff when it raises, up to max_attempts times. A TimeoutError, including a node's own timeout_ms, fails the node without a retry. Orchestrator.execute and Swarm.execute catch their own errors and return a result with success=False, so a node that wraps one should check result.success and raise if you want the retry to fire. An orchestrator still reports success=True when a single specialist fails: that failure is on result.specialist_results[<id>].error, so check it as well if a failed specialist should trigger the retry. Pick the shape that fits each layer of the problem.

See also

  • Agent Loop — the loop every agent in every shape runs.
  • Hooks — observe and steer across all of them.
  • Streaming — the typed event taxonomy.