DEV Community: Deepti Reddy

You Decide: Build a Human-Directed Editorial Workflow with the Manual Strategy

Deepti Reddy — Fri, 01 May 2026 11:22:37 +0000

By Deepti Reddy | April 2026

This is Part 7 of an 8-part series covering every multi-agent strategy in AgentSpan. Today: the manual strategy — the human picks which agent speaks next. Full control.

In Part 1, we built a sequential pipeline where each agent’s output feeds the next. In Part 2, we built a parallel code review where three reviewers running simultaneously. In Part 3, the LLM decided which agent runs. In Part 4, agents transferred between each other peer-to-peer. In Part 5, agents took turns in a fixed rotation: architect, security, pragmatist, architect, security, pragmatist. Predictable and structured. In Part 6, agents were selected randomly.

Every strategy so far has one thing in common: something other than the human decided which agent runs next — the pipeline, the LLM, the classifier, the rotation, or randomness.

That is the manual strategy. The human picks. Every turn, the workflow pauses and asks: which agent should go next? The human makes the call.

What is AgentSpan

AgentSpan is an orchestration layer for building, bringing, and observing AI agents as durable workflows.

Build : define agents with the AgentSpan SDK using Agent, @tool, and 8 multi-agent strategies. Compiles to server-side workflows that survive crashes.
Bring : already using an agent framework such as LangGraph, OpenAI Agents SDK, or Google ADK? Pass your agents directly to run(). AgentSpan adds durability and orchestration on top.
Observe : every execution is inspectable in the dashboard. See agent flows, inputs/outputs, tool calls, and token usage. Debug failures, replay runs.

Setup

Two commands:

pip install agentspan
agentspan server start

This gives you a local AgentSpan server with a visual dashboard at localhost:6767.

What we are building

An editorial workflow with three specialists:

Writer : drafts and revises content
Fact Checker : verifies accuracy, flags unsupported claims
Copy Editor : fixes grammar, tightens prose, improves flow

The human editor directs the workflow. Maybe the writer goes first, then the fact checker, then back to the writer for revisions, then the copy editor for polish. The human decides the sequence based on what the draft needs — not a predetermined pipeline.

Human picks: writer

-> [Writer] drafts content

Human picks: fact_checker

-> [Fact Checker] flags issues

Human picks: writer

-> [Writer] revises based on feedback

Human picks: copy_editor

-> [Copy Editor] polishes final version

How is this different from everything else?

Manual is the only strategy where a human is in the orchestration loop — not just approving tool calls (that is HITL), but actively directing which agent runs.

Defining the specialists

Three agents with editorial roles:

from agentspan.agents import Agent, AgentRuntime, Strategy, EventType

writer = Agent(
name="writer",
model="openai/gpt-4o",
instructions=(
"You are a writer. Expand on ideas with clear, engaging prose. "
"If you receive feedback from other agents, revise your work "
"based on their suggestions. Keep your response focused and concise."
),
)

fact_checker = Agent(
name="fact_checker",
model="openai/gpt-4o",
instructions=(
"You are a fact checker. Review the content for accuracy. "
"Flag any claims that are unsupported, exaggerated, or wrong. "
"Be specific - quote the exact text and explain the issue. "
"If everything checks out, say so."
),
)

copy_editor = Agent(
name="copy_editor",
model="openai/gpt-4o",
instructions=(
"You are a copy editor. Review the content for grammar, clarity, "
"tone, and flow. Suggest specific edits. Tighten prose. Remove "
"filler. Make it read well. Return the improved version."
),
)

Each agent sees the full conversation — including what other agents said before them. The writer can revise based on the fact checker’s feedback. The copy editor polishes after everyone else is done.

The manual strategy

team = Agent(name="editorial_team",
model="openai/gpt-4o",
agents=[writer, fact_checker, copy_editor],
strategy=Strategy.MANUAL,
max_turns=4,
)

No router. No classifier. No rotation. The workflow pauses at each turn and waits for the human to pick.
Compare all eight strategies:

# Sequential: fixed order, each runs once
pipeline = a >> b >> c

# Parallel: all run at once
team = Agent(agents=[a, b, c], strategy=Strategy.PARALLEL)

# Handoff: parent LLM picks one
triage = Agent(agents=[a, b, c], strategy=Strategy.HANDOFF)

# Router: classifier picks one
triage = Agent(agents=[a, b, c], strategy=Strategy.ROUTER, router=classifier)

# Swarm: agents transfer between each other
team = Agent(agents=[a, b, c], strategy=Strategy.SWARM)

# Round robin: fixed rotation
debate = Agent(agents=[a, b, c], strategy=Strategy.ROUND_ROBIN, max_turns=6)

# Random: random selection each turn
brainstorm = Agent(agents=[a, b, c], strategy=Strategy.RANDOM, max_turns=6)

# Manual: human picks each turn
workflow = Agent(agents=[a, b, c], strategy=Strategy.MANUAL, max_turns=4)

Same Agent class. Eight strategies. One primitive.

Running it

with AgentRuntime() as runtime:handle = runtime.start(
team, "Write a short paragraph about the history of artificial intelligence."
)

for event in handle.stream():
if event.type == EventType.WAITING:
print("Pick the next agent:")
choice = input("> ")
handle.respond({"selected": choice})
elif event.type == EventType.DONE:
print(event.output)

The workflow starts. Each turn, EventType.WAITING fires and the human types an agent name. handle.respond() sends the selection back to the server. The chosen agent runs, and the cycle repeats until max_turns.

Example session

Pick the next agent: writer, fact_checker, copy_editor

> writer
[Writer produces a draft about AI history]
Pick the next agent: writer, fact_checker, copy_editor

> fact_checker
[Fact checker reviews, flags one claim about the Dartmouth Conference date]
Pick the next agent: writer, fact_checker, copy_editor

> writer
[Writer revises the draft, fixes the flagged claim]
Pick the next agent: writer, fact_checker, copy_editor
> copy_editor
[Copy editor polishes grammar, tightens prose, returns final version]

The human directed the flow: write, check, revise, polish. A fixed pipeline could not do this — the human decided to send it back to the writer after the fact checker flagged an issue. That loop was a judgment call, not a predetermined step.

When to use manual

Manual is the right strategy when:

Editorial workflows — human editors direct the review process based on what the content needs
Training and education — a student directs which tutor agent to consult next
Debugging — a developer manually directs diagnostic agents to investigate specific areas
Quality gates — a human reviewer decides whether the output needs another pass or is ready

If the routing logic can be automated with rules, LLM judgment, or fixed order, use one of the other seven strategies. Manual is for when the human’s judgment is the routing logic.

How durability works

Same as every previous part. The workflow pauses durably on the server while waiting for the human’s selection. Kill the process. Come back an hour later. The workflow is still waiting. Respond via the CLI or reconnect from a new process:

agentspan agent respond <execution-id> — data ‘{“selected”: “writer”}’

No timeout. No thread blocked. The server holds the state indefinitely.

What’s next

Part 8: Router — A dedicated classifier agent decides which specialist runs — not the parent LLM, not a fixed order, but a separate lightweight model whose only job is routing. The final strategy in the series.

Try it

pip install agentspan
agentspan server start
python 07_editorial_manual.py

GitHub : github.com/agentspan-ai/agentspan
Blog examples : github.com/agentspan-ai/agentspan/tree/main/sdk/python/examples/blog_and_videos/manual
Docs : agentspan.ai/docs
Discord : https://discord.com/invite/ajcA66JcKq

Roll the Dice: Build a Brainstorming Session with the Random Strategy

Deepti Reddy — Fri, 01 May 2026 11:06:32 +0000

By Deepti Reddy | April 2026

This is Part 6 of an 8-part series covering every multi-agent strategy in AgentSpan. Today, we will cover the random strategy, a random agent is selected each turn. No rotation, no decision — pure randomness.

But what if predictability is the problem? In brainstorming, you do not want a fixed order. You want surprise. You want the creative thinker to jump in twice in a row, or the critical thinker to challenge an idea the moment it lands. Fixed rotations produce fixed thinking.

That is the random strategy. Each turn, a random agent is selected. No pattern. No schedule. The same agent might go twice in a row, or not at all for three turns. The randomness creates variety in perspective that a fixed rotation cannot.

What is AgentSpan

AgentSpan is an orchestration layer for building, bringing, and observing AI agents as durable workflows.

Build : define agents with the AgentSpan SDK using Agent, @tool, and 8 multi-agent strategies. Compiles to server-side workflows that survive crashes.
Bring : already using an agent framework such as LangGraph, OpenAI Agents SDK, or Google ADK? Pass your agents directly to run(). AgentSpan adds durability and orchestration on top.
Observe : every execution is inspectable in the dashboard. See agent flows, inputs/outputs, tool calls, and token usage. Debug failures, replay runs.

Setup

Two commands:

pip install agentspan
agentspan server start

This gives you a local AgentSpan server with a visual dashboard at localhost:6767.

What we are building

A brainstorming session with three thinkers:

Creative : bold, unconventional ideas — “what if we did the opposite?”
Practical : feasibility, timelines, resources — “what would it actually take?”
Critical : risks, holes, blind spots — “what could go wrong?”

Each turn, one is randomly selected. After 6 turns, a summarizer distills the session into the top ideas and next steps.

Turn 1: [Creative] <- random
Turn 2: [Critical] <- random
Turn 3: [Creative] <- random (again!)
Turn 4: [Practical] <- random
Turn 5: [Critical] <- random
Turn 6: [Practical] <- random
[Summarizer] -> top 3 ideas + next step

How is this different from round robin?

In round robin , the order is fixed: A, B, C, A, B, C. Every agent gets equal time. Every agent knows when their turn is.

In random , there is no order. Agent A might speak three times. Agent C might speak once. The distribution is uneven by design — some perspectives naturally dominate in any real brainstorming session, and that is fine.

Defining the thinkers

Three agents with deliberately different thinking styles:

from agentspan.agents import Agent, AgentRuntime, Strategy

creative = Agent(
name="creative",
model="openai/gpt-4o",
instructions=(
"You are a creative thinker. Generate bold, unconventional ideas. "
"Push boundaries. Think 'what if we did the opposite of what everyone "
"expects?' Read what others said before you and build on their ideas "
"or take them in a surprising direction. Keep your response to 2–3 paragraphs."
),
)

practical = Agent(
name="practical",
model="openai/gpt-4o",
instructions=(
"You are a practical thinker. Focus on what can actually be built "
"and shipped. Consider timelines, resources, and feasibility. Read "
"what others said before you and ground their ideas in reality - "
"what would it take to actually do this? Keep your response to 2–3 paragraphs."
),
)

critical = Agent(
name="critical",
model="openai/gpt-4o",
instructions=(
"You are a critical thinker. Find the holes, the risks, the things "
"nobody wants to talk about. Read what others said before you and "
"stress-test their ideas - what could go wrong? What are they not "
"considering? Keep your response to 2–3 paragraphs."
),
)

The key instruction in each: “Read what others said before you.” Each agent sees the full conversation. The difference from round robin is that who speaks next is random, not predetermined.

The summarizer

After the brainstorm, a summarizer produces actionable output:

summarizer = Agent(name="summarizer",
model="openai/gpt-4o",
instructions=(
"You observed a brainstorming session between a creative thinker, "
"a practical thinker, and a critical thinker. Produce a summary:\n\n"
"1. Top 3 ideas (ranked by potential)\n"
"2. Biggest risk identified\n"
"3. Recommended next step (one sentence)\n\n"
"Be concise and actionable."
),
)

The random strategy

brainstorm = Agent(name="brainstorm",
model="openai/gpt-4o",
agents=[creative, practical, critical],
strategy=Strategy.RANDOM,
max_turns=6,
)
pipeline = brainstorm >> summarizer

max_turns=6 means 6 randomly selected turns. Some agents might go multiple times, others might be skipped entirely. Then >> pipes the brainstorm transcript to the summarizer.

Compare the strategies:

# Sequential: fixed order, each runs once
pipeline = a >> b >> c

# Parallel: all run at once
team = Agent(agents=[a, b, c], strategy=Strategy.PARALLEL)

# Handoff: parent LLM picks one
triage = Agent(agents=[a, b, c], strategy=Strategy.HANDOFF)

# Router: classifier picks one
triage = Agent(agents=[a, b, c], strategy=Strategy.ROUTER, router=classifier)

# Swarm: agents transfer between each other
team = Agent(agents=[a, b, c], strategy=Strategy.SWARM)

# Round robin: fixed rotation
debate = Agent(agents=[a, b, c], strategy=Strategy.ROUND_ROBIN, max_turns=6)

# Random: random selection each turn
brainstorm = Agent(agents=[a, b, c], strategy=Strategy.RANDOM, max_turns=6)

Same Agent class. Different strategy. Different behavior.

Running it

with AgentRuntime() as runtime:result = runtime.run(
pipeline,
"How should a developer tools company get its first 1,000 users?",
)
result.print_result()

Every run produces a different conversation because the agent selection is random. Run it twice and you get two different brainstorming sessions. That is the point.

When to use random

Random is not a strategy for everything. It is specifically useful when:

Brainstorming — you want diverse, unpredictable perspectives
Load balancing across models — distribute prompts across GPT-4o, Claude, Gemini randomly to compare output quality
Stress testing — randomly select different scenarios to hit an agent with
Creative writing — different voices or styles contribute randomly to a collaborative piece

If you need every agent to participate equally, use round robin. If you need one specific agent, use handoff or router. Random is for when variety itself is the goal.

How durability works

The random selection is made server-side and persisted. If your process crashes after turn 4:

Turns 1–4 and their random selections are persisted on the server.
You restart your script.
Turns 5–6 continue with new random selections — the first 4 are not re-run.

Composability

Random composes with other strategies:

# Random brainstorm, then structured review
brainstorm = Agent(agents=[creative, practical, critical], strategy=Strategy.RANDOM, max_turns=6)
review = Agent(agents=[architect, security], strategy=Strategy.ROUND_ROBIN, max_turns=4)
pipeline = brainstorm >> review >> summarizer

Random generates ideas. Round robin reviews them. The summarizer produces the final output. Three strategies, one pipeline.

Try it

pip install agentspan
agentspan server start

python 07_brainstorm_random.py

GitHub : github.com/agentspan-ai/agentspan
Blog examples : github.com/agentspan-ai/agentspan/tree/main/sdk/python/examples/blog_and_videos/random
Docs : agentspan.ai/docs
Discord : https://discord.com/invite/ajcA66JcKq

What’s next

Part 7: Manual — The human picks which agent speaks next. No LLM deciding, no classifier, no randomness — full human control over the orchestration. The final strategy in the series.

Three Reviewers, Two Rounds, One Verdict: Build a Code Review Debate with Round Robin

Deepti Reddy — Fri, 24 Apr 2026 18:49:43 +0000

By Deepti Reddy | April 2026

This is Part 5 of an 8-part series covering every multi-agent strategy in AgentSpan. Today, we will cover the round robin strategy where agents take turns in a fixed rotation, each building on what the others said.

— -

What is AgentSpan

AgentSpan is an orchestration layer for building, bringing, and observing AI agents as durable workflows.

Build: define agents with the AgentSpan SDK using Agent, @tool, and 8 multi-agent strategies. Compiles to server-side workflows that survive crashes.
Bring: already using an agent framework such as LangGraph, OpenAI Agents SDK, or Google ADK? Pass your agents directly to run(). AgentSpan adds durability and orchestration on top.
Observe: every execution is inspectable in the dashboard. See agent flows, inputs/outputs, tool calls, and token usage. Debug failures, replay runs.

Setup

Two commands:

pip install agentspan
agentspan server start

This gives you a local AgentSpan server with a visual dashboard at localhost:6767.

— -

Every strategy so far either runs agents once or lets someone (a parent, a classifier, or the agents themselves) decide who goes next. But what about tasks where you want every agent to weigh in, then weigh in again — a back-and-forth discussion where each round builds on the last?

That is the round robin strategy. No routing. No decisions about who goes next. Just a fixed rotation: A, B, C, A, B, C. Each agent sees the full conversation history — what every other agent said before them — and builds on it.

What we are building

A code review debate with three reviewers who take turns:

Architect: focuses on design, structure, scalability
Security Reviewer: focuses on vulnerabilities, data exposure, insecure defaults
Pragmatist: focuses on what actually matters for shipping — minimum fix, what can wait

They review the same code. But unlike parallel (Part 2), where each reviewer works independently, round robin reviewers read each other’s comments and respond to them. The architect raises a design concern. The security reviewer agrees and adds a vulnerability. The pragmatist pushes back — “that is a follow-up PR, not a blocker.” Two rounds of this, then a summarizer produces the verdict.

Round 1: [Architect] -> [Security] -> [Pragmatist]
Round 2: [Architect] -> [Security] -> [Pragmatist]
 |
 [Summarizer] -> APPROVE / REQUEST CHANGES

How is this different from parallel?

In parallel (Part 2), all three reviewers run at the same time on the same input. They never see each other’s work. You get three independent opinions.

In round robin, reviewers take turns. Each one sees the full conversation — including what previous reviewers said. The second-round architect can respond to what the pragmatist said in round one. It is a discussion, not three monologues.

Defining the reviewers

Three agents, each with a different perspective:

architect = Agent(
 name=”architect”,
 model=”openai/gpt-4o”,
 instructions=(
 “You are a software architect reviewing code. Focus on:\n”
 “- Design patterns and structure\n”
 “- Separation of concerns\n”
 “- Scalability and maintainability\n\n”
 “Read what other reviewers said before you. Build on their points, “
 “don’t repeat them. Keep your response to 2–3 paragraphs.”
 ),
)

security_reviewer = Agent(
 name=”security_reviewer”,
 model=”openai/gpt-4o”,
 instructions=(
 “You are a security engineer reviewing code. Focus on:\n”
 “- Injection vulnerabilities (SQL, command, path traversal)\n”
 “- Authentication and authorization gaps\n”
 “- Data exposure and insecure defaults\n\n”
 “Read what other reviewers said before you. Build on their points, “
 “don’t repeat them. Keep your response to 2–3 paragraphs.”
 ),
)

pragmatist = Agent(
 name=”pragmatist”,
 model=”openai/gpt-4o”,
 instructions=(
 “You are a senior engineer who values shipping. Focus on:\n”
 “- Is this good enough to merge today?\n”
 “- What is the minimum fix needed?\n”
 “- What can wait for a follow-up PR?\n\n”
 “Push back on over-engineering. Read what other reviewers said “
 “and decide what actually matters for this PR. “
 “Keep your response to 2–3 paragraphs.”
 ),
)

The key instruction in each: “Read what other reviewers said before you. Build on their points, don’t repeat them.” This is what makes round robin a discussion, not a repetition. Each agent knows it is part of a conversation.

The summarizer

After the debate, a summarizer reads the entire transcript and produces a verdict:

summarizer = Agent(
 name=”summarizer”,
 model=”openai/gpt-4o”,
 instructions=(
 “You observed a code review discussion between an architect, “
 “a security reviewer, and a pragmatist. Produce a final verdict:\n\n”
 “1. APPROVE, REQUEST CHANGES, or NEEDS DISCUSSION\n”
 “2. Must-fix items (block merge)\n”
 “3. Nice-to-have items (follow-up PR)\n”
 “4. One-sentence summary\n\n”
 “Be decisive. Don’t hedge.”
 ),
)

The round robin strategy

review = Agent(
 name=”code_review_round_robin”,
 model=”openai/gpt-4o”,
 agents=[architect, security_reviewer, pragmatist],
 strategy=Strategy.ROUND_ROBIN,
 max_turns=6,
)

pipeline = review >> summarizer

max_turns=6 means 6 total turns: architect, security, pragmatist, architect, security, pragmatist. Two full rounds. Then >> pipes the discussion transcript to the summarizer for a verdict.

Compare all six strategies:

# Sequential — fixed order, each runs once
pipeline = a >> b >> c

# Parallel — all run at once
team = Agent(agents=[a, b, c], strategy=Strategy.PARALLEL)

# Handoff — parent LLM picks one
triage = Agent(agents=[a, b, c], strategy=Strategy.HANDOFF)

# Router — classifier picks one
triage = Agent(agents=[a, b, c], strategy=Strategy.ROUTER, router=classifier)

# Swarm — agents transfer between each other
team = Agent(agents=[a, b, c], strategy=Strategy.SWARM)

# Round Robin — fixed rotation, agents respond to each other
debate = Agent(agents=[a, b, c], strategy=Strategy.ROUND_ROBIN, max_turns=6)

Same Agent class. Different strategy. Different behavior.

The code to review

A short Python snippet with a mix of design, security, and pragmatic issues:

import sqlite3
import os

def get_user(db_path, user_id):
 conn = sqlite3.connect(db_path)
 query = f"SELECT * FROM users WHERE id = {user_id}"
 result = conn.execute(query).fetchone()
 conn.close()
 return result

def save_upload(filename, data):
 path = f"/uploads/{filename}"
 with open(path, "wb") as f:
 f.write(data)
 os.chmod(path, 0o777)
 return path

def process_payment(amount, card_number):
 print(f"Processing ${amount} on card {card_number}")
 return {"status": "ok", "amount": amount}

Three functions. SQL injection. Path traversal. Card number in logs. File permissions wide open. No error handling. But also — it is short, it works, and someone wants to merge it. The three reviewers will have very different opinions about what to do.

Running it

with AgentRuntime() as runtime:
 result = runtime.run(pipeline, code)
 result.print_result()

What happens

Round 1:

The architect goes first. Sees the raw code. Flags the lack of abstraction — no database layer, no upload service, direct file system access. Suggests separating concerns.

The security reviewer goes second. Sees the architect’s comments AND the code. Agrees on the design issues, then adds the critical security findings: SQL injection in get_user, path traversal in save_upload, card number logged in plaintext by process_payment, 0o777 permissions.

The pragmatist goes third. Sees both previous reviews. Agrees the SQL injection and card logging are blockers — those must be fixed before merge. But pushes back on the architecture refactoring: “That is a follow-up PR. The code works. Fix the security issues, merge, and refactor next sprint.”

Round 2:

The architect responds to the pragmatist: concedes the full refactor can wait, but insists on at minimum extracting the database connection into a context manager.

The security reviewer doubles down: the path traversal and 0o777 are just as critical as the SQL injection. All three security issues must be fixed, not just the SQL one.

The pragmatist agrees on all three security fixes, accepts the context manager suggestion, and calls it: “Fix those four things and this is ready to merge.”

Summarizer reads the whole transcript and produces the verdict.

This is the value of round robin over parallel. In parallel (Part 2), you get three independent opinions and have to synthesize them yourself. In round robin, the agents synthesize for you — they debate, converge, and the summarizer captures the consensus.

Round Robin vs Parallel: when to use which

Use parallel when speed matters and independence is a feature. Use round robin when convergence matters and you want agents to challenge each other.

How durability works

Same as every previous part. The round robin compiles into a durable DoWhile loop on the server. If your process crashes after turn 4 (second round, security reviewer just finished):

Turns 1–4 are persisted on the server.
You restart your script.
Turn 5 (pragmatist, round 2) resumes — no re-running the first 4 turns.

Each turn in the rotation is a durable checkpoint.

Composability

Round robin composes naturally:

# Debate, then summarize
pipeline = review >> summarizer

# Parallel fan-out for data, then round robin debate on findings
research = Agent(agents=[market, technical, financial], strategy=Strategy.PARALLEL)
debate = Agent(agents=[optimist, skeptic], strategy=Strategy.ROUND_ROBIN, max_turns=4)

pipeline = research >> debate >> summarizer

Parallel collects independent research. Round robin debates the findings. Summarizer produces the final recommendation. Three strategies, one pipeline.

Try it

pip install agentspan
agentspan server start
python 06_code_review_debate.py

GitHub: github.com/agentspan-ai/agentspan
Blog examples: github.com/agentspan-ai/agentspan/tree/main/sdk/python/examples/blog_and_videos/round_robin
Docs: agentspan.ai/docs
Discord: https://discord.com/invite/ajcA66JcKq

What’s next

Part 6: Random — A random agent is selected each turn. Not a rotation, not a decision but pure randomness. Useful for creative brainstorming, load balancing across models, and generating diverse output ensembles. Same Agent class, different strategy.

Agents Talk to Each Other: Build a Support Swarm with Peer-to-Peer Transfers

Deepti Reddy — Fri, 24 Apr 2026 07:05:51 +0000

By Deepti Reddy | April 2026

This is Part 4 of an 8-part series covering every multi-agent strategy in AgentSpan. Today: the swarm strategy agents hand off to each other using transfer tools. No manager and no classifier. Peer-to-peer.

— -

In Part 1, we built a sequential pipeline, agents in a fixed order. In Part 2, we built a parallel code review agents running simultaneously. In Part 3, we built a handoff triage bot where the parent LLM decides the route.

Every strategy so far has been top-down. A parent agent, a classifier, or a fixed pipeline decides which agent runs. The agents themselves never talk to each other. The bug handler in Part 3 never said “actually, this should go to the docs handler.”

That is the swarm strategy. There is no parent deciding. Instead, every agent gets auto-generated transfer_to_ tools. The refund specialist can transfer to tech support. Tech support can transfer back to the refund specialist. They decide among themselves — peer-to-peer.

What we are building

A customer support system with three agents that can transfer between each other:

Front-line Support: triages incoming requests, looks up orders
Refund Specialist: checks eligibility, processes refunds
Tech Support: checks warranty, creates warranty claims

The front-line agent gets the initial message. When it recognizes the issue, it transfers to the right specialist. But here is the key difference from handoff: the specialists can transfer to each other. If a customer asks for a refund but the issue is actually a product defect, the refund specialist transfers to tech support — no need to go back to the front-line agent first.

Customer → [Front-line] → transfer_to_refund_specialist()
 ↓
 [Refund Specialist] → “this is a defect, not a return”
 ↓
 transfer_to_tech_support()
 ↓
 [Tech Support] → warranty claim created

How is this different from handoff and router?

In handoff (Part 3), the parent agent’s LLM decides everything. Sub-agents are tools the parent calls. Once the bug handler runs, it cannot say “actually, send this to the docs handler”, it finishes and control returns to the parent.

In router, a dedicated classifier picks one specialist. Same limitation: once the specialist runs, it cannot redirect.

In swarm, every agent has transfer tools for every other agent. The LLM inside each agent can decide to transfer at any point during its own execution. There is no parent in the loop.

Setup

Same as Parts 1 through 3:

pip install agentspan
agentspan server start

Defining the specialists

Each specialist has tools for its domain:

from agentspan.agents import Agent, AgentRuntime, Strategy, tool

from agentspan.agents.handoff import OnTextMention

@tool
def lookup_order(order_id: str) -> dict:
 “””Look up order details and status.”””
 return {
 “order_id”: order_id,
 “status”: “delivered”,
 “item”: “Wireless Headphones (Model WH-1000)”,
 “amount”: 249.99,
 }


@tool
def check_refund_eligibility(order_id: str) -> dict:
 “””Check if an order is eligible for a refund.”””
 return {
 “order_id”: order_id,
 “eligible”: True,
 “refund_amount”: 249.99,
 “processing_time”: “3–5 business days”,
 }


@tool
def process_refund(order_id: str, amount: float, reason: str) -> dict:
 “””Process a refund for an order.”””
 return {
 “refund_id”: “RF-88431”,
 “amount”: amount,
 “status”: “processed”,
 }


@tool
def check_warranty(order_id: str) -> dict:
 “””Check warranty status for a product.”””
 return {
 “order_id”: order_id,
 “warranty_status”: “active”,
 “warranty_expiry”: “2027–04–10”,
 “claim_options”: [“replacement”, “repair”],
 }


@tool
def create_warranty_claim(order_id: str, issue: str) -> dict:
 “””Create a warranty claim.”””
 return {
 “claim_id”: “WC-55102”,
 “next_step”: “Prepaid shipping label via email within 24 hours”,
 }

Now the agents. Notice the instructions — each specialist knows when to transfer to the other:

refund_specialist = Agent(
 name=”refund_specialist”,
 model=”openai/gpt-4o”,
 instructions=(
 “You are a refund specialist. Handle refund requests.\n\n”
 “1. Use check_refund_eligibility to verify the order qualifies.\n”
 “2. If eligible, use process_refund to issue the refund.\n”
 “3. Confirm the refund amount, method, and timeline.\n\n”
 “If the customer’s issue is actually a product defect (not a return), “
 “transfer to tech_support — they handle warranty claims.”
 ),
 tools=[check_refund_eligibility, process_refund],
)

tech_support = Agent(
 name=”tech_support”,
 model=”openai/gpt-4o”,
 instructions=(
 “You are technical support. Handle product issues and warranty claims.\n\n”
 “1. Use check_warranty to verify warranty status.\n”
 “2. If under warranty, use create_warranty_claim.\n”
 “3. Explain next steps to the customer.\n\n”
 “If the customer just wants their money back (not a replacement/repair), “
 “transfer to refund_specialist.”
 ),
 tools=[check_warranty, create_warranty_claim],
)

The refund specialist knows: “if this is a defect, transfer to tech_support.” Tech support knows: “if they just want money back, transfer to refund_specialist.” They can bounce the customer between themselves — no parent agent involved.

The swarm strategy

support = Agent(
 name=”support”,
 model=”openai/gpt-4o”,
 instructions=(
 “You are front-line customer support. Triage the request.\n”
 “- Refund or return → transfer to refund_specialist\n”
 “- Product issue or defect → transfer to tech_support\n”
 “Do NOT handle refunds or technical issues yourself.”
 ),
 agents=[refund_specialist, tech_support],
 strategy=Strategy.SWARM,
 tools=[lookup_order],
 handoffs=[
 OnTextMention(text=”refund”, target=”refund_specialist”),
 OnTextMention(text=”defect”, target=”tech_support”),
 ],
 max_turns=5,
)

Two things are happening here:

Auto-generated transfer tools (primary mechanism). When you set strategy=Strategy.SWARM, AgentSpan automatically creates transfer_to_refund_specialist() and transfer_to_tech_support() as callable tools on the front-line agent. The LLM sees these tools and decides when to call them. Each specialist also gets transfer tools for every other agent, so refund_specialist gets transfer_to_tech_support() and vice versa.
OnTextMention fallback conditions (optional safety net). The handoffs= list provides fallback rules: if the agent’s response mentions “refund” but it did not call a transfer tool, the condition triggers and forces the handoff. These are optional — you can omit handoffs= entirely and rely purely on the auto-generated transfer tools.

The transfer tools are the core mechanism. The OnTextMention conditions are belt-and-suspenders.

Compare all five strategies:

# Sequential — fixed order, all run
pipeline = a >> b >> c

# Parallel — all run at once
team = Agent(agents=[a, b, c], strategy=Strategy.PARALLEL)

# Handoff — parent LLM picks one
triage = Agent(agents=[a, b, c], strategy=Strategy.HANDOFF)

# Router — separate classifier picks one
triage = Agent(agents=[a, b, c], strategy=Strategy.ROUTER, router=classifier)

# Swarm — agents transfer between each other
team = Agent(agents=[a, b, c], strategy=Strategy.SWARM)

Same Agent class. Different strategy. Different behavior.

Running it

with AgentRuntime() as runtime:
 result = runtime.run(
 support,
 “I bought wireless headphones (order ORD-7821) last week and “
 “the left ear cup stopped working after 3 days. I want my money back.”,
 )
 result.print_result()

This message is interesting because it has two intents: a product defect (left ear cup stopped working) AND a refund request (I want my money back). Watch what happens.

The flow

Front-line reads the message, sees “money back” → calls transfer_to_refund_specialist()
Refund specialist reads the full context, sees “stopped working after 3 days” → recognizes this is a defect, not a simple return → calls transfer_to_tech_support()
Tech support checks warranty (active until 2027), creates a warranty claim, offers replacement or repair options

That re-routing refund specialist deciding “actually, this is a tech issue” and transferring — is the thing that handoff and router cannot do. In those strategies, once a specialist runs, it finishes. In a swarm, agents redirect mid-conversation.

The transfer tool mechanism

When you set strategy=Strategy.SWARM, AgentSpan auto-generates transfer tools. You do not define them. They appear as regular tools that the LLM can call.

For the front-line agent:

Available tools:

— lookup_order(order_id)

— transfer_to_refund_specialist() ← auto-generated

— transfer_to_tech_support() ← auto-generated

For the refund specialist:

Available tools:

— check_refund_eligibility(order_id)

— process_refund(order_id, amount, reason)

— transfer_to_tech_support() ← auto-generated

— transfer_to_support() ← auto-generated (back to front-line)

Every agent can reach every other agent. The LLM decides when — based on its instructions and the conversation context.

OnTextMention: the fallback

The handoffs= conditions are optional. They exist as a safety net:

handoffs=[
 OnTextMention(text=”refund”, target=”refund_specialist”),
 OnTextMention(text=”defect”, target=”tech_support”),
]

The evaluation order is:

Did the LLM call a transfer tool? → Yes → use that transfer. Done.
No transfer tool called? → Check OnTextMention conditions against the agent’s response text.
Condition matches? → Force the handoff.
Nothing matches? → No transfer. Agent’s turn ends.

The transfer tools are priority 1. The text conditions are priority 2. You can omit handoffs= entirely and the swarm still works — the LLM-driven transfers are the primary mechanism.

When to use swarm vs handoff vs router

Handoff

Use when: You want a parent agent to delegate.

Pattern: Top-down. Parent picks a child, child runs, control returns.

Example: Issue triage — read once, route once.

Router

Use when: Classification is straightforward and you want cost control.

Pattern: Top-down. Classifier picks, specialist runs.

Example: Ticket categorization at scale — cheap model classifies.

Swarm

Use when: Conversations evolve and agents need to redirect.

Pattern: Peer-to-peer. Any agent can transfer to any other.

Example: Support conversations where the initial classification is wrong.

The mental model: handoff and router are dispatchers, they pick once. Swarm is a team conversation as agents pass the baton as the situation evolves.

How durability works

Same as every previous part. The swarm compiles into a durable server-side workflow. If your process crashes after the refund specialist transfers to tech support but before tech support finishes:

The transfer decision is persisted on the server.
You restart your script.
Tech support resumes from where it was, no re-triage, no re-transfer.

Every transfer is a durable state transition on the server.

Composability

Swarm composes with other strategies:

# Swarm handles the conversation, then a summarizer wraps up
team = Agent(
 agents=[support, refund_specialist, tech_support],
 strategy=Strategy.SWARM,
)

pipeline = team >> summarizer

Or use a swarm inside a larger handoff:

# Handoff decides: simple question → single agent, complex issue → swarm team
support_swarm = Agent(
 agents=[refund_specialist, tech_support],
 strategy=Strategy.SWARM,
)

router = Agent(
 agents=[faq_agent, support_swarm],
 strategy=Strategy.HANDOFF,
)

Same Agent class. Strategies nest naturally.

Try it

pip install agentspan
agentspan server start
python 05_support_swarm.py

GitHub: github.com/agentspan-ai/agentspan
Blog examples: github.com/agentspan-ai/agentspan/tree/main/sdk/python/examples/blog_and_videos/swarm
Docs: agentspan.ai/docs
Discord: https://discord.com/invite/ajcA66JcKq

What’s next

Part 5: Round Robin — Agents take turns in a fixed rotation. No routing decision at all, just A, B, C, A, B, C. Useful for debates, iterative refinement, and multi-perspective review. Same Agent class, different strategy.

The LLM Decides: Build an Issue Triage Bot with the Handoff Strategy

Deepti Reddy — Mon, 13 Apr 2026 07:28:52 +0000

By Deepti Reddy | April 2026

This is Part 3 of an 8-part series covering every multi-agent strategy in AgentSpan. Today, we will go over the handoff strategy — the LLM decides which agent handles the task.

In Part 1, we built a sequential pipeline: triage >> root_cause >> postmortem. Each agent’s output feeds the next. In Part 2, we built a parallel code review: three reviewers run simultaneously on the same code.

Both of those patterns have a fixed structure. You know upfront which agents run and in what order. But what about tasks where you don’t know the path until you see the input?

A GitHub issue comes in. It might be a bug. It might be a feature request. It might be a docs question. You don’t know until you read it and each type needs a completely different handler.

That is the handoff strategy. The LLM reads the input, decides which specialist should handle it, and hands off. One agent runs, not all of them.

What we are building

An issue triage bot with three specialist agents:

Bug Handler : analyzes severity, affected component, repro steps
Feature Handler : summarizes the request, estimates complexity
Docs Handler : identifies the gap, drafts a reply

The triage agent reads the issue and picks one. Not a fixed pipeline. Not fan-out. The LLM decides the route.

Issue → [Triage] → reads it → decides → [Bug Handler]
or [Feature Handler]
or [Docs Handler]

Why not sequential or parallel?

Sequential (>>) runs every agent in order. But a feature request doesn’t need the bug handler. You would waste API calls running agents that have nothing to do.

Parallel runs every agent at the same time. But you don’t need three opinions on the same issue , you need the right one.

Handoff runs exactly one agent — the one the LLM picks. It’s a router that thinks, not a pipeline that grinds.

Setup

Same as Part 1 and Part 2:

pip install agentspan
agentspan server start

Defining the specialists

Three agents, each with tight instructions that prevent hallucination:

from agentspan.agents import Agent, AgentRuntime, Strategy

bug_handler = Agent(
name="bug_handler",
model="openai/gpt-4o",
instructions=(
"You handle bug reports. Read the issue CAREFULLY.\n\n"
"You MUST output EXACTLY this format and NOTHING else - no greeting, "
"no explanation, no sign-off, no extra text:\n\n"
"Severity: P0/P1/P2/P3\n"
"Component: <which part>\n"
"Repro steps: <what the user described, or 'Not provided'>\n"
"Labels: bug, <severity>\n"
"Engineering summary: <2–3 sentences>\n\n"
"Example output:\n"
"Severity: P2\n"
"Component: REST API - /users endpoint\n"
"Repro steps: Send a GET request with limit=0. Returns 500 instead of empty list.\n"
"Labels: bug, P2\n"
"Engineering summary: Off-by-one in pagination. The /users endpoint does not "
"handle limit=0. Affects v2.1+ only.\n\n"
"RULES:\n"
"- ONLY use information the user actually wrote. No guesses.\n"
"- Do NOT suggest workarounds or fixes.\n"
"- Do NOT add any text outside the format."
),
)

feature_handler = Agent(
name="feature_handler",
model="openai/gpt-4o",
instructions=(
"You handle feature requests. Read the issue CAREFULLY.\n\n"
"You MUST output EXACTLY this format and NOTHING else - no greeting, "
"no explanation, no sign-off, no extra text:\n\n"
"Request: <one sentence>\n"
"Use case: <in the user's words, or 'No use case provided'>\n"
"Complexity: small/medium/large\n"
"Labels: enhancement, <area>\n"
"Community summary: <2–3 sentences>\n\n"
"Example output:\n"
"Request: Add CSV export for agent execution history.\n"
"Use case: User wants to import execution data into their BI tool for "
"weekly reporting.\n"
"Complexity: small\n"
"Labels: enhancement, observability\n"
"Community summary: Request for CSV export of execution history. User "
"needs it for BI/reporting integration. Low complexity - the data is "
"already queryable.\n\n"
"RULES:\n"
"- ONLY use information the user actually wrote. No guesses.\n"
"- Do NOT promise timelines or delivery.\n"
"- Do NOT add any text outside the format."
),
)

docs_handler = Agent(
name="docs_handler",
model="openai/gpt-4o",
instructions=(
"You handle docs issues and questions. Read the issue CAREFULLY.\n\n"
"You MUST output EXACTLY this format and NOTHING else - no greeting, "
"no explanation, no sign-off, no extra text:\n\n"
"Confusion: <what the user is stuck on>\n"
"Doc gap: <which doc page is missing or unclear>\n"
"Draft reply: <under 50 words - acknowledge the gap and say the "
"team will update the docs>\n"
"Labels: documentation\n\n"
"Example output:\n"
"Confusion: User doesn't know how to configure retry behavior.\n"
"Doc gap: The tools page does not mention retry configuration.\n"
"Draft reply: Good catch - the docs don't cover this yet. "
"We'll add a section on retry configuration to the tools page.\n"
"Labels: documentation\n\n"
"RULES:\n"
"- ONLY describe the gap. Do NOT answer the technical question.\n"
"- NEVER write code examples - you don't have access to the "
"source code and will get it wrong.\n"
"- Keep Draft reply under 50 words. Just acknowledge and commit "
"to updating the docs.\n"
"- Do NOT add any text outside the format."
),
)

Same Agent class as Parts 1 and 2 and same pattern. The difference is what comes next.

The handoff strategy

triage = Agent(name="triage",
model="openai/gpt-4o",
agents=[bug_handler, feature_handler, docs_handler],
strategy=Strategy.HANDOFF,
instructions=(
"You are an issue triage bot. Your ONLY job is to route.\n\n"
"1. Read the issue.\n"
"2. Hand off to exactly ONE agent:\n"
" - Error/crash/traceback/regression → bug_handler\n"
" - Feature request/suggestion → feature_handler\n"
" - Docs question/confusion → docs_handler\n"
"3. After the specialist responds, output their response "
"VERBATIM. Copy-paste it exactly. Add nothing.\n\n"
"You are a router, not an analyst. Do NOT add your own words."
),
)

The triage agent has access to three sub-agents. It reads the input, picks one, and delegates. Compare to the other strategies:

Sequential — fixed order, all run

pipeline = bug_handler >> feature_handler >> docs_handler

Parallel — all run at once

team = Agent(agents=[bug_handler, feature_handler, docs_handler],
strategy=Strategy.PARALLEL)

Handoff — LLM picks one

triage = Agent(agents=[bug_handler, feature_handler, docs_handler],
strategy=Strategy.HANDOFF)

Same agents. Different strategy. Different behavior.

Running it

with AgentRuntime() as runtime:
  result = runtime.run(
  triage,
  "File upload fails with a 500 error when the filename has spaces. "
  "Uploading 'report.pdf' works, but 'Q1 report.pdf' returns a server "
  "error. Looks like the filename isn't being URL-encoded.",
  )

result.print_result()

Result → bug_handler

The triage agent reads it, sees “fails with a 500 error” and “works / doesn’t work.” Routes to bug_handler:

Severity: P1
Component: File Upload Service
Repro steps: Upload a file with spaces in the filename like
'Q1 report.pdf'. Returns a 500 error.
Labels: bug, P1
Engineering summary: The File Upload Service fails to handle
filenames with spaces, leading to a 500 server error. Likely
a URL encoding issue with filenames containing spaces.

No invented details. Repro steps are exactly what the user said. No workarounds suggested.

Now swap the input to a feature request:

result = runtime.run(triage,
"Would be great to have a drag-and-drop zone for file uploads "
"instead of the file picker button. Most modern apps support this.",
)

Routes to feature_handler. Swap it to a docs question:
result = runtime.run(
triage,
"The docs mention a max file size limit but don't say what it is "
"or how to change it. Where do I configure this?",
)

Routes to docs_handler. Same triage bot, same code. The LLM decides the path each time.

How durability works

Same as Parts 1 and 2 — the handoff compiles into a durable workflow on the server. If your process crashes after the triage agent picks bug_handler but before bug_handler finishes:

The triage decision is persisted on the server.
You restart your script.
bug_handler resumes — no re-triage, no re-reading the issue.

This matters when you are triaging a backlog of 50 issues overnight. If the process dies at issue 37, it picks up at 37 — not back to 1.

Going further: connect to GitHub + Discord

The hardcoded issues were fine for learning Strategy.HANDOFF. But in production, you’d fetch real issues from GitHub and route notifications to Discord. Here’s how.

Store your credentials

In the AgentSpan dashboard (localhost:6767), go to Credentials and add:

Name: GITHUB_TOKEN
Value: Your GitHub personal access token (needs repo scope)

Name: DISCORD_TOKEN
Value: Discord bot token

Discord setup

Go to discord.com/developers/applications → New Application
Bot tab → Reset Token → copy it
Turn on Message Content Intent
OAuth2 → URL Generator → select bot scope → permissions: Send Messages, Read Message History, Add Reactions
Open the generated URL to invite the bot to your server
Create channels: #bugs, #feature-requests, #docs
Copy each channel ID (right-click channel → Copy Channel ID)

The tools

from agentspan.agents import tool
import os
import requests

GITHUB_API = "https://api.github.com"
DISCORD_API = "https://discord.com/api/v10"
DISCORD_CHANNELS = {
"bugs": "YOUR_BUGS_CHANNEL_ID",
"feature-requests": "YOUR_FEATURES_CHANNEL_ID",
"docs": "YOUR_DOCS_CHANNEL_ID",
}

@tool(credentials=["GITHUB_TOKEN"])
def get_issue(repo: str, issue_number: int) -> dict:
"""Fetch a GitHub issue by number."""
token = os.environ["GITHUB_TOKEN"]
resp = requests.get(
f"{GITHUB_API}/repos/{repo}/issues/{issue_number}",
headers={"Authorization": f"Bearer {token}"},
)

issue = resp.json()
return {
"number": issue["number"],
"title": issue["title"],
"body": issue.get("body", ""),
"user": issue["user"]["login"],
"labels": [l["name"] for l in issue.get("labels", [])],
}

@tool(credentials=["GITHUB_TOKEN"])
def search_issues(repo: str, query: str) -> list:
"""Search for similar or duplicate issues."""
token = os.environ["GITHUB_TOKEN"]
resp = requests.get(
f"{GITHUB_API}/search/issues",
headers={"Authorization": f"Bearer {token}"},
params={"q": f"{query} repo:{repo}", "per_page": 5},
)
return [
{"number": i["number"], "title": i["title"], "state": i["state"]}
for i in resp.json().get("items", [])
]

@tool(credentials=["GITHUB_TOKEN"])
def add_labels(repo: str, issue_number: int, labels: list) -> dict:
"""Add labels to a GitHub issue."""
token = os.environ["GITHUB_TOKEN"]
requests.post(
f"{GITHUB_API}/repos/{repo}/issues/{issue_number}/labels",
headers={"Authorization": f"Bearer {token}"},
json={"labels": labels},
)
return {"status": "labeled", "labels": labels}

@tool(credentials=["GITHUB_TOKEN"])
def post_comment(repo: str, issue_number: int, body: str) -> dict:
"""Post a comment on a GitHub issue."""
token = os.environ["GITHUB_TOKEN"]
requests.post(
f"{GITHUB_API}/repos/{repo}/issues/{issue_number}/comments",
headers={"Authorization": f"Bearer {token}"},
json={"body": body},
)
return {"status": "commented", "issue_number": issue_number}

@tool(credentials=["DISCORD_TOKEN"])
def post_to_discord(channel_name: str, message: str) -> dict:
"""Post a message to a Discord channel. channel_name: bugs, feature-requests, or docs."""
token = os.environ["DISCORD_TOKEN"]
channel_id = DISCORD_CHANNELS.get(channel_name)
if not channel_id:
return {"status": "skipped", "reason": f"No channel configured for #{channel_name}"}
requests.post(
f"{DISCORD_API}/channels/{channel_id}/messages",
headers={"Authorization": f"Bot {token}"},
json={"content": message},
)
return {"status": "posted", "channel": channel_name}

Wire them in

The specialists gain tools — they label the issue, post a comment, and notify the right Discord channel. We also add a fetcher agent that pulls the issue from GitHub first, then hands off to the triage handoff:

fetcher = Agent(name="fetcher",
model="openai/gpt-4o",
instructions=(
"You fetch GitHub issues. Call get_issue with the repo and "
"issue_number from the prompt. Return the issue's title and "
"body verbatim."
),
tools=[get_issue],
)

bug_handler = Agent(
name="bug_handler",
model="openai/gpt-4o",
instructions=(
"You handle bug reports. Read the issue CAREFULLY.\n\n"
"Do these steps in order:\n"
"1. Search for duplicate issues.\n"
"2. Add labels: 'bug' + severity label.\n"
"3. Post an analysis comment on the GitHub issue.\n"
"4. Post a summary to the 'bugs' Discord channel.\n\n"
"Do NOT guess details the user didn't provide."
),
tools=[search_issues, add_labels, post_comment, post_to_discord],
)

feature_handler = Agent(
name="feature_handler",
model="openai/gpt-4o",
instructions=(
"You handle feature requests. Read the issue CAREFULLY.\n\n"
"Do these steps in order:\n"
"1. Search for duplicate requests.\n"
"2. Add labels: 'enhancement' + area label.\n"
"3. Post a comment acknowledging the request.\n"
"4. Post to 'feature-requests' Discord channel.\n\n"
"Do NOT promise timelines. Do NOT invent use cases."
),
tools=[search_issues, add_labels, post_comment, post_to_discord],
)

docs_handler = Agent(
name="docs_handler",
model="openai/gpt-4o",
instructions=(
"You handle docs issues. Read the issue CAREFULLY.\n\n"
"Do these steps in order:\n"
"1. Add the label 'documentation'.\n"
"2. Post a reply on the GitHub issue under 100 words.\n"
"3. Post to 'docs' Discord channel.\n\n"
"If you don't know the answer, say 'The team will follow up.'"
),
tools=[add_labels, post_comment, post_to_discord],
)

triage = Agent(
name="triage",
agents=[bug_handler, feature_handler, docs_handler],
strategy=Strategy.HANDOFF,
instructions=(
"Read the issue. Hand off to ONE agent:\n"
"- Error/crash/traceback → bug_handler\n"
"- Feature request/suggestion → feature_handler\n"
"- Docs question/confusion → docs_handler\n\n"
"Do NOT answer it yourself. Just route."
),
)

Sequential: fetcher pulls the issue, then triage routes it

pipeline = fetcher >> triage

Run it against a real FastAPI issue:

The fetcher agent has one job: call get_issue to pull the issue from GitHub and return its title and body. That is why we chain it sequentially before triage (fetcher >> triage) — so the fetched issue flows straight into the handoff. Without it, you wouldhave to paste the raw issue text into the prompt yourself, like the hardcoded example.

with AgentRuntime() as runtime:

result = runtime.run( pipeline, "Triage issue #13056 in repo fastapi/fastapi")

result.print_result()

The fetcher pulls the issue from GitHub. The triage agent reads it, hands off to the right specialist. The specialist labels the issue, posts a comment, and pings the right Discord channel. Full loop — GitHub in, GitHub + Discord out.

Try it

pip install agentspan

agentspan server start

python 03_issue_triage_handoff.py

# For the GitHub + Discord integration version:
# 1. Add GITHUB_TOKEN and DISCORD_TOKEN in the AgentSpan dashboard (Credentials page)
# 2. Update REPO, ISSUE_NUMBER, and DISCORD_CHANNELS in 03_issue_triage_github_discord.py
# 3. python 03_issue_triage_github_discord.py

GitHub : github.com/agentspan/agentspan
Blog examples : github.com/agentspan/agentspan/tree/main/sdk/python/examples/blog_and_videos/handoff
Docs : agentspan.ai/docs
Discord: https://discord.com/invite/ajcA66JcKq

What’s next

Part 4: The Router Strategy — Like handoff, but with a dedicated classifier agent that handles the routing decision. Separate the “who should handle this” from the “handle it” — cheaper, faster, more controllable. Same Agent class, different strategy.

Three Reviewers, One PR, Zero Wait: Build a Parallel AI Agent Pipeline

Deepti Reddy — Fri, 10 Apr 2026 04:12:48 +0000

By Deepti Reddy | April 2026

This is Part 2 of an 8-part series covering every multi-agent strategy in AgentSpan. Today: the parallel strategy — multiple agents tackle the same input simultaneously.

In Part 1, we built a sequential pipeline where each agent’s output feeds into the next: classifier >> responder >> escalation_checker. That works when steps depend on each other.

But what about tasks where the steps are independent? A code review doesn’t need the bug reviewer to finish before the security reviewer starts. They can all look at the same code at the same time.

Today, I will show you how to build a parallel code review pipeline with AgentSpan. Three reviewers, one piece of code, all running simultaneously. Results arrive together — no waiting.

What is Agentspan

AgentSpan is an orchestration layer for building, bringing, and observing AI agents as durable workflows.

Build: define agents with the Agentspan SDK using Agent, @tool, and 8 multi-agent strategies. Compiles to server-side workflows that survive crashes.
Bring: already using an agent framework such as LangGraph, OpenAI Agents SDK, or Google ADK. Pass your agents directly to run(). AgentSpan adds durability and orchestration on top.
Observe: every execution is inspectable in the dashboard. See agent flows, inputs/outputs, tool calls, and token usage. Debug failures, replay runs.

What we are building

A parallel code review with three specialist agents:

Bug Reviewer : finds logic errors, crashes, and unhandled edge cases
Security Reviewer : finds injection flaws, insecure defaults, and OWASP issues
Style Reviewer : checks for type hints, docstrings, naming, and code quality

All three receive the same code. All three run at the same time. Their results are collected into sub_results — one per agent.

Code → [Bug Reviewer] → bug findings
→ [Security Reviewer] → security findings → combined review
→ [Style Reviewer] → style suggestions
This is the parallel strategy - agents fan out on the same input and their results arrive together.

Setup

Same as Part 1:

pip install agentspan
agentspan server start

Defining the agents

Three agents, each with a different specialty:

from agentspan.agents import Agent, AgentRuntime, Strategy

bug_reviewer = Agent(
name="bug_reviewer",
model="openai/gpt-4o",
instructions=(
"You are a senior software engineer reviewing code for bugs. "
"Read the ACTUAL code carefully. Only report issues you can "
"point to in specific lines. Quote the exact code and explain "
"what's wrong.\n\n"
"Look for: logic errors, unhandled edge cases, crashes, and "
"incorrect behavior.\n\n"
"If the code has no bugs, say 'No bugs found.' "
"Do NOT invent issues. Do NOT give generic advice."
),
)

security_reviewer = Agent(
name="security_reviewer",
model="openai/gpt-4o",
instructions=(
"You are an application security engineer reviewing code for "
"vulnerabilities. Read the ACTUAL code carefully. Only report "
"vulnerabilities you can point to in specific lines.\n\n"
"Look for: injection flaws, insecure defaults, data exposure, "
"missing input validation, OWASP Top 10 issues. Rate each "
"finding as Critical, High, Medium, or Low.\n\n"
"If the code has no security issues, say 'No security issues "
"found.' Do NOT invent vulnerabilities. Do NOT give generic "
"security advice."
),
)

style_reviewer = Agent(
name="style_reviewer",
model="openai/gpt-4o",
instructions=(
"You are a Python code quality reviewer. Read the ACTUAL code "
"carefully. Only report style issues you can point to in "
"specific lines.\n\n"
"Look for: missing type hints, missing docstrings, hardcoded "
"values, print vs logging, naming issues, readability.\n\n"
"If the code style is good, say 'Code style looks good.' "
"Do NOT invent issues. Do NOT give generic advice."
),
)

Same Agent class as Part 1. Same model, same instructions pattern. The difference is what comes next.

The parallel strategy

Instead of >>, we wrap the agents in a parent agent with Strategy.PARALLEL:

review = Agent(name="code_review",
model="openai/gpt-4o",
agents=[bug_reviewer, security_reviewer, style_reviewer],
strategy=Strategy.PARALLEL,
)

All three agents will run simultaneously on the same input. Compare this to sequential:

Sequential — each waits for the previous one

pipeline = bug_reviewer >> security_reviewer >> style_reviewer

Parallel — all run at the same time

review = Agent(name="code_review",
model="openai/gpt-4o",
agents=[bug_reviewer, security_reviewer, style_reviewer],
strategy=Strategy.PARALLEL,
)

Same agents, different strategy. The sequential version takes 3x as long because each reviewer waits for the one before it. The parallel version takes as long as the slowest reviewer — roughly 1x.

The code to review

Let us feed it a short Python script with a mix of bugs, security issues, and style problems:

code = """                                                                                                     
  import os                                                                                                      

  def process_upload(filename, data):                                                                            
      path = f"/uploads/{filename}"                                                                              
      with open(path, "wb") as f:                                                                                
          f.write(data)                                     
      os.chmod(path, 0o777)
      return path                                                                                                

  def get_user(db, user_id):                                                                                     
      query = f"SELECT * FROM users WHERE id = {user_id}"   
      return db.execute(query).fetchone()     

  def send_welcome(user):
      print(f"Welcome {user['name']}!")                                                                          
      return True
  """

Three functions, each with problems that a different reviewer would catch. The bug reviewer, security reviewer, and style reviewer each have something to find — but they don’t need each other’s output to do their job.

Running the review

with AgentRuntime() as runtime:result = runtime.run(review, code)
for agent_name, sub in result.sub_results.items():
print(f"\n{'=' * 50}")
print(f" {agent_name}")
print(f"{'=' * 50}")
print(sub)

The key difference from Part 1: result.sub_results. In a sequential pipeline, you get one final output. In a parallel pipeline, you get a dictionary, one entry per agent.

Let us see what each reviewer found.

Bug Reviewer

1. Security Risk (Path Injection): In the process_upload function,
the line path = f"/uploads/{filename}" can be exploited if
filename contains directory traversal characters (e.g.,
../../etc/passwd). This could allow an attacker to write or
overwrite protected files on the system. Proper validation or
sanitization of filename is necessary to prevent this.

2. Security Risk (SQL Injection): In the get_user function, the
query query = f"SELECT * FROM users WHERE id = {user_id}"
directly interpolates user_id into the SQL query string. If
user_id is not properly sanitized, this code is vulnerable to
SQL injection attacks. Using parameterized queries or prepared
statements would mitigate this risk.
No other bugs found.

Security Reviewer

1. Filename Injection: The process_upload function constructs a file
path using the unsanitized filename parameter directly from user
input. This can lead to path traversal vulnerabilities. For
example, a filename like ../../etc/passwd could be used to write
outside the intended directory. Rating: High
Reference: path = f"/uploads/{filename}"

2. Insecure File Permissions: The file uploaded through
process_upload is given permissions 777, which means it is
readable, writable, and executable by all users. This poses a
security risk as anyone could modify or execute the file.
Rating: High
Reference: os.chmod(path, 0o777)

3. SQL Injection: The get_user function constructs an SQL query by
directly embedding the user_id parameter into the query string
without any sanitization. This can lead to SQL injection if
user_id is derived from user-controlled input. Rating: Critical
Reference: query = f"SELECT * FROM users WHERE id = {user_id}"
No other security issues found.

Style Reviewer

1. Missing Type Hints:
- process_upload is missing type hints for its parameters and
return value.
- get_user is missing type hints for its parameters and return
value.
- send_welcome is missing type hints for its parameter and
return value.
2. Missing Docstrings:
- process_upload lacks a docstring explaining its purpose.
- get_user lacks a docstring explaining its purpose.
- send_welcome lacks a docstring explaining its purpose.
3. Hardcoded Values:
- In process_upload, the path "/uploads/" and the permissions
0o777 are hardcoded.
4. Print vs Logging:
- send_welcome uses print() to output text, which is generally
discouraged in favor of using a logging framework.

Three different perspectives, all from the same code, all produced simultaneously. The bug reviewer found path injection and SQL injection, then said “no other bugs found.” The security reviewer rated each vulnerability by severity and quoted the exact lines. The style reviewer focused on code quality without inventing issues. Each brought a unique lens — and where they overlap (SQL injection), their analysis differs.

Sequential vs Parallel: when to use which

Sequential (>>)                                                                                                
  • Use when: Each step depends on the previous                                                                  
  • Execution: One at a time, in order                                                                           
  • Speed: Sum of all agent times                                                                                
  • Output: Single final result                                                                                  
  • Example: Classify → respond → escalate                                                                       

  Parallel (Strategy.PARALLEL)                                                                                   
  • Use when: Steps are independent                                                                              
  • Execution: All at once                                  
  • Speed: Time of the slowest agent                                                                             
  • Output: sub_results dict, one per agent                 
  • Example: Bug check + security check + style check

The rule is simple: if agent B needs agent A’s output, use sequential. If they can work independently, use parallel.

How durability works (same as sequential)

The parallel pipeline compiles into a durable server-side workflow — same as the sequential pipeline from Part 1. If your process crashes while two reviewers are done and one is still running:

The two completed results are persisted on the server.
You restart your script.
Only the third reviewer re-runs.
Results are collected and returned.

No wasted API calls. No re-running agents that already finished.

Composability

You can mix parallel and sequential in the same pipeline:

# Run all three reviews in parallel, then summarize sequentially
review = Agent(
name="code_review",
model="openai/gpt-4o",
agents=[bug_reviewer, security_reviewer, style_reviewer],
strategy=Strategy.PARALLEL,
)

summarizer = Agent(
name="summarizer",
model="openai/gpt-4o",
instructions="Combine the three reviews into a single verdict: APPROVE, REQUEST CHANGES, or NEEDS DISCUSSION.",
)

pipeline = review >> summarizer

Parallel fan-out, then sequential summarization. Same Agent class, same >> operator, strategies compose naturally.

The full code

from agentspan.agents import Agent, AgentRuntime, Strategy

bug_reviewer = Agent(name="bug_reviewer",
model="openai/gpt-4o",
instructions=(
"You are a senior software engineer reviewing code for bugs. "
"Read the ACTUAL code carefully. Only report issues you can "
"point to in specific lines. Quote the exact code and explain "
"what's wrong.\n\n"
"Look for: logic errors, unhandled edge cases, crashes, and "
"incorrect behavior.\n\n"
"If the code has no bugs, say 'No bugs found.' "
"Do NOT invent issues. Do NOT give generic advice."
),
)

security_reviewer = Agent(
name="security_reviewer",
model="openai/gpt-4o",
instructions=(
"You are an application security engineer reviewing code for "
"vulnerabilities. Read the ACTUAL code carefully. Only report "
"vulnerabilities you can point to in specific lines.\n\n"
"Look for: injection flaws, insecure defaults, data exposure, "
"missing input validation, OWASP Top 10 issues. Rate each "
"finding as Critical, High, Medium, or Low.\n\n"
"If the code has no security issues, say 'No security issues "
"found.' Do NOT invent vulnerabilities. Do NOT give generic "
"security advice."
),
)

style_reviewer = Agent(
name="style_reviewer",
model="openai/gpt-4o",
instructions=(
"You are a Python code quality reviewer. Read the ACTUAL code "
"carefully. Only report style issues you can point to in "
"specific lines.\n\n"
"Look for: missing type hints, missing docstrings, hardcoded "
"values, print vs logging, naming issues, readability.\n\n"
"If the code style is good, say 'Code style looks good.' "
"Do NOT invent issues. Do NOT give generic advice."
),
)

review = Agent(
name="code_review",
model="openai/gpt-4o",
agents=[bug_reviewer, security_reviewer, style_reviewer],
strategy=Strategy.PARALLEL,
)

code = """
Review this code:
import os
def process_upload(filename, data):
path = f"/uploads/{filename}"
with open(path, "wb") as f:
f.write(data)
os.chmod(path, 0o777)
return path
def get_user(db, user_id):
query = f"SELECT * FROM users WHERE id = {user_id}"
return db.execute(query).fetchone()
def send_welcome(user):
print(f"Welcome {user['name']}!")
return True
"""

with AgentRuntime() as runtime:
result = runtime.run(review, code)
for agent_name, sub in result.sub_results.items():
print(f"\n{'=' * 50}")
print(f" {agent_name}")
print(f"{'=' * 50}")
print(sub)

Going further: connect to GitHub

The hardcoded code snippet was fine for learning Strategy.PARALLEL. But in production, you would pull a real PR diff from GitHub and post the review back. Here is how.

Store your credentials

In the AgentSpan dashboard (localhost:6767), go to Credentials and add:

Name: GITHUB_TOKEN
Value: Your GitHub personal access token (needs repo scope)

The tools

Two tools — one to fetch the PR diff, one to post the review:

from agentspan.agents import tool
import os
import requests

GITHUB_API = "https://api.github.com"

@tool(credentials=["GITHUB_TOKEN"])
def get_pr_diff(repo: str, pr_number: int) -> dict:
"""Fetch the diff for a GitHub pull request. repo format: owner/repo"""
token = os.environ["GITHUB_TOKEN"]
resp = requests.get(
f"{GITHUB_API}/repos/{repo}/pulls/{pr_number}",
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github.v3.diff",
},
)

pr_info = requests.get(
f"{GITHUB_API}/repos/{repo}/pulls/{pr_number}",
headers={"Authorization": f"Bearer {token}"},
).json()

return {
"title": pr_info.get("title", ""),
"description": pr_info.get("body", ""),
"diff": resp.text[:10000],
"files_changed": pr_info.get("changed_files", 0),
"additions": pr_info.get("additions", 0),
"deletions": pr_info.get("deletions", 0),
}

@tool(credentials=["GITHUB_TOKEN"])
def post_pr_review(repo: str, pr_number: int, body: str) -> dict:
"""Post a review comment on a GitHub pull request."""
token = os.environ["GITHUB_TOKEN"]
resp = requests.post(
f"{GITHUB_API}/repos/{repo}/pulls/{pr_number}/reviews",
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github.v3+json",
},
json={"body": body, "event": "COMMENT"},
)

return {"status": "posted", "pr_number": pr_number}

Wire them in

The three reviewers stay the same — no tools needed. Add a fetcher to pull the PR diff and a summarizer that combines the reviews and posts to GitHub:

fetcher = Agent(name="pr_fetcher",
model="openai/gpt-4o",
instructions=(
"You are a helper that fetches PR diffs. Call the get_pr_diff "
"tool and return the COMPLETE diff verbatim as your output."
),
tools=[get_pr_diff],
)

summarizer = Agent(
name="summarizer",
model="openai/gpt-4o",
instructions=(
"You are a tech lead. Given three code review outputs (bugs, "
"security, style), combine them into a single PR review comment "
"in markdown with sections: ## Bugs, ## Security, ## Style, "
"## Verdict (APPROVE / REQUEST CHANGES / NEEDS DISCUSSION)."
),
)

# Sequential: fetch diff → parallel review → summarize
pipeline = fetcher >> review >> summarizer

The parallel review runs the same as before — three agents, same code, simultaneous. The fetcher pulls the real PR diff from GitHub, the reviewers analyze it, and the summarizer combines everything into one review.

That isthe full loop: fetch PR → three parallel reviews → summarize → post review comment.

Try it

pip install agentspan
agentspan server start
python 02_code_review_parallel.py

# For the GitHub integration version:
# 1. Add GITHUB_TOKEN in the AgentSpan dashboard (Credentials page)
# 2. Update REPO and PR_NUMBER in 02_code_review_parallel_github.py
# to point at your own repo and PR
# 3. python 02_code_review_parallel_github.py

GitHub : github.com/agentspan/agentspan
Blog examples : github.com/agentspan/agentspan/tree/main/sdk/python/examples/blog_and_videos/parallel_execution
Docs : agentspan.ai/docs
Discord: https://discord.com/invite/ajcA66JcKq

What’s next

Part 3: The Handoff Strategy — The LLM decides which agent to delegate to. A router that thinks, not just matches. Same Agent class, different strategy.

From Support Ticket to Resolution: Build a Sequential AI Agent Pipeline

Deepti Reddy — Wed, 08 Apr 2026 20:28:02 +0000

By Deepti Reddy | April 2026

This is Part 1 of an 8-part series covering every multi-agent strategy in AgentSpan. Today, we will cover the sequential pipeline with the >> operator.

What is Agentspan

AgentSpan is an orchestration layer for building, bringing, and observing AI agents as durable workflows.

Build: define agents with the Agentspan SDK using Agent, @tool, and 8 multi-agent strategies. Compiles to server-side workflows that survive crashes.
Bring: already using an agent framework such as LangGraph, OpenAI Agents SDK, or Google ADK. Pass your agents directly to run(). AgentSpan adds durability and orchestration on top.
Observe: every execution is inspectable in the dashboard. See agent flows, inputs/outputs, tool calls, and token usage. Debug failures, replay runs.

To get started with Agentspan, download and install the Python library.

pip install agentspan
agentspan server start

What we are building

Your support queue has 47 tickets. Half are urgent. Your team is triaging by gut feel, writing the same “we are looking into it” response and forgetting to escalate the ones that actually need engineering.

What if every ticket got classified, responded to, and escalation-checked in seconds consistently, every time?

Today, I will show you how to build a support triage pipeline with AgentSpan. Three agents, one operator: classify the ticket, draft a response, and decide if engineering needs to get involved. The whole thing runs in seconds, and if your process crashes mid-pipeline, it picks up exactly where it left off.

A support triage pipeline with three specialist agents:

Classifier : reads the ticket and tags priority, category, product area, sentiment, and key facts
Responder : drafts a reply to post as a comment on the ticket
Escalation Checker : decides whether engineering needs to be pulled in, and writes an internal note on the ticket if so

Each agent’s output feeds into the next. The classifier informs the responder. The responder and classifier together inform the escalation checker decision.

Ticket → [Classifier] → structured tags → [Responder] → draft reply → [Escalation Checker] → escalate or not

This is the sequential strategy — agents run one after another, each building on the previous agent’s work. In AgentSpan, you express this with a single operator: >>.

Setup

Two commands:

pip install agentspan
agentspan server start

This gives you a local AgentSpan server with a visual dashboard at localhost:6767.

Defining the agents

Each agent is an instance of the Agent class with a name, a model, and instructions that define its expertise:

from agentspan.agents import Agent, AgentRuntime

classifier = Agent(
name="classifier",
model="openai/gpt-4o",
instructions=(
"You are a support ticket classifier. Given a customer support "
"ticket, produce a structured classification:\n"
"- Customer: name and company (from the ticket)\n"
"- Priority: P0 (outage), P1 (critical), P2 (degraded), P3 (minor)\n"
"- Category: performance, bug, feature-request, billing, onboarding\n"
"- Product area: which part of the product is affected\n"
"- Customer sentiment: frustrated, neutral, positive\n"
"- Key facts: bullet list of the specific issues mentioned\n\n"
"Be concise and structured. Do not guess - only classify based "
"on what the customer actually wrote."
),
)

responder = Agent(
name="responder",
model="openai/gpt-4o",
instructions=(
"You are a senior support engineer drafting a response to a "
"customer. Given the original ticket and the classification, "
"write a professional reply that:\n"
"- Acknowledges their issue and urgency\n"
"- Confirms what you understand the problem to be\n"
"- Asks specific follow-up questions if anything is unclear\n"
"- Sets expectations for next steps and timeline\n\n"
"Be empathetic but not generic. Reference the specific details "
"they mentioned. Keep it under 200 words."
),
)

escalation_checker = Agent(
name="escalation_checker",
model="openai/gpt-4o",
instructions=(
"You are a support team lead reviewing a ticket for escalation. "
"Given the original ticket, classification, and draft response, "
"decide:\n"
"1. Should this be escalated to engineering? (yes/no)\n"
"2. Why or why not? (one sentence)\n"
"3. If yes, write an internal note for the engineering team - "
"include the customer impact, urgency, and what to investigate.\n"
"4. If no, confirm the support team can handle it and why.\n\n"
"Be direct. Engineers are busy - give them only what they need."
),
)

Each agent has one job. The classifier doesn’t try to write a response. The responder doesn’t decide on escalation. Separation of concerns — same principle you would use in code.

No tools needed here as these agents are pure LLM reasoning. Their instructions are their expertise.

The >> operator

Here is the entire orchestration logic:

pipeline = classifier >> responder >> escalation_checker

The >> operator chains agents sequentially. The output of each agent becomes the input to the next.

Under the hood, AgentSpan compiles this into a durable server-side workflow. It is equivalent to writing:

from agentspan.agents import Strategy

pipeline = Agent(
name="support_triage",
agents=[classifier, responder, escalation_checker],
strategy=Strategy.SEQUENTIAL,
)

The >> operator is shorthand — cleaner to read, identical behavior. Use whichever you prefer.

The ticket

Let’s feed it a realistic support ticket:

ticket = “””
Subject: URGENT - API extremely slow, blocking our monthly batch processing
From: David Park <david.park@acmelogistics.com>
Company: Acme Logistics (Enterprise plan)
Environment: Production (US-East)
Submitted: Monday 3:15 PM EST
Hi Support Team,
We're in the middle of our monthly batch processing and we're completely
stuck. This is our most critical operational window of the month.
Here's what's happening:
1. Our batch API calls are going through, but response times have gone
from ~200ms to 15–30 SECONDS per call. We have 50,000+ items to
process and at this rate it will take days instead of hours.
2. The web dashboard is nearly unusable - pages take 45+ seconds to
load, and we keep getting timeout errors when trying to view our
job status.
3. We tried splitting our batch into smaller chunks thinking it was a
rate limit issue, but even individual API calls are slow.
4. Nothing changed on our end - same code, same volume as last month
when everything ran fine in under 2 hours.
We have downstream systems waiting on this data and our SLA with our
own customers is at risk. Three of our team members have been stuck
on this since 1 PM and can't do anything else until it's resolved.
Can someone please look into this ASAP? We need to know:
- Is there a known issue on your end?
- Is there anything we can do to work around it?
- When can we expect normal performance?
Thanks,
David Park
Senior Platform Engineer, Acme Logistics
"""

This is a real-world ticket — an enterprise customer blocked on critical operations, detailed symptoms, clear asks. The kind of ticket that needs fast, accurate handling.

Running the pipeline

with AgentRuntime() as runtime:
result = runtime.run(pipeline, ticket)
result.print_result()

AgentRuntime handles everything from compiling the pipeline into a workflow, registering it with the server, starting worker processes, and streaming the result back. One function call.

Let us walk through what each agent produces.

Step 1: Classifier output

The classifier reads the ticket and produces a structured assessment:

- Customer: David Park, Acme Logistics
- Priority: P0 (outage)
- Category: performance
- Product area: API and web dashboard
- Customer sentiment: frustrated
- Key facts:
  - API response times increased from ~200ms to 15–30 seconds.
  - Monthly batch processing is critical and currently blocked.
  - 50,000+ items to process; delays are significant, extending from hours to potential days.
  - Web dashboard is very slow, with page load times of 45+ seconds and frequent timeout errors.
  - Previous code and volume worked without issues last month.
  - Significant impact on internal operations and SLAs with their own customers.Category: performance
  - Product area: API and web dashboard
  - Customer sentiment: frustrated
- Key facts:
  - API response times increased from ~200ms to 15–30 seconds.
  - Monthly batch processing is critical and currently blocked.
  - 50,000+ items to process; delays are significant, extending from hours to potential days.
  - Web dashboard is very slow, with page load times of 45+ seconds and frequent timeout errors.
  - Previous code and volume worked without issues last month.
  - Significant impact on internal operations and SLAs with their own customers.

Notice what it did: extracted the customer name, classified the priority, tagged the right product areas, and pulled every specific detail from the ticket. This structured output is exactly what the responder needs.

Step 2: Responder output

The responder receives the classification (not the raw ticket) and drafts a customer-facing reply:

Hello David,

Thank you for reaching out and I sincerely apologize for the disruption
you're experiencing. I understand the urgency and the impact this has on
your operations and SLAs with your customers.
It seems that both the API and web dashboard have experienced a significant
degradation in performance, with API response times increasing from ~200ms
to 15–30 seconds and the web dashboard taking over 45 seconds to load.
This challenge is particularly critical as it is blocking your monthly
batch processing of over 50,000 items, risking extending your processing
timeline significantly.
To better assist you, could you confirm if there have been any configuration
changes or software updates on your end since last month? Additionally, are
there specific times when the slowdown is more pronounced, or is it
consistent throughout the day?
Our team is prioritizing this issue and has started an investigation. I will
provide an update within the next hour and continue to keep you informed as
we progress. Thank you for your patience as we work towards resolving this
matter swiftly.

Best regards,
[Your Name]
Senior Support Engineer

It addresses David by name, references his exact details of the 200ms-to-30s degradation, the 50K items, the SLA risk. It asks targeted follow-up questions and commits to an update within the hour.

Step 3: Escalation Checker output

The escalation checker reviews everything and makes the call:

1. Should this be escalated to engineering? Yes.
2. Why or why not? The significant performance degradation in both
the API and web dashboard indicates a potential underlying issue
that requires technical investigation by engineering.
3. Internal Note for Engineering: The customer is experiencing a
drastic increase in API response times from ~200ms to 15–30 seconds
and web dashboard load times exceeding 45 seconds. This is
critically impacting their monthly batch processing of 50,000
items, threatening their SLAs. Please investigate any potential
backend issues contributing to this performance degradation
urgently.

Direct and actionable. The engineering team gets the customer impact, the numbers, and what to investigate.

In production, the responder’s output would be posted as a public comment on the ticket — the customer sees it in their thread. The escalation checker’s internal note would be posted as a private note on the same ticket where only your team sees it. We will wire that up in the Zendesk section below. For now, we are just looking at the output.

Three agents, three distinct outputs, each building on the last. The whole pipeline ran in under 30 seconds.

How durability works

When you call runtime.run(), AgentSpan doesn’t just chain three API calls in your Python process. It compiles the >> pipeline into a durable workflow on the server. The server owns the execution state — your Python process is just a client.

Here is what that means in practice:

The classifier finishes. Its output is persisted on the server.
Your laptop crashes.
You restart your script. AgentSpan reconnects to the same execution.
The pipeline resumes at the responder — the classification is still there.
No re-processing, no wasted API calls, no lost work.

This is the core difference between AgentSpan and chaining API calls in a script. Your support tooling should be as reliable as the SLAs you are trying to meet.

The full code

from agentspan.agents import Agent, AgentRuntime

classifier = Agent(
name="classifier",
model="openai/gpt-4o",
instructions=(
"You are a support ticket classifier. Given a customer support "
"ticket, produce a structured classification:\n"
"- Customer: name and company (from the ticket)\n"
"- Priority: P0 (outage), P1 (critical), P2 (degraded), P3 (minor)\n"
"- Category: performance, bug, feature-request, billing, onboarding\n"
"- Product area: which part of the product is affected\n"
"- Customer sentiment: frustrated, neutral, positive\n"
"- Key facts: bullet list of the specific issues mentioned\n\n"
"Be concise and structured. Do not guess - only classify based "
"on what the customer actually wrote."
),
)

responder = Agent(
name="responder",
model="openai/gpt-4o",
instructions=(
"You are a senior support engineer drafting a response to a "
"customer. Given the original ticket and the classification, "
"write a professional reply that:\n"
"- Acknowledges their issue and urgency\n"
"- Confirms what you understand the problem to be\n"
"- Asks specific follow-up questions if anything is unclear\n"
"- Sets expectations for next steps and timeline\n\n"
"Be empathetic but not generic. Reference the specific details "
"they mentioned. Keep it under 200 words."
),
)

escalation_checker = Agent(
name="escalation_checker",
model="openai/gpt-4o",
instructions=(
"You are a support team lead reviewing a ticket for escalation. "
"Given the original ticket, classification, and draft response, "
"decide:\n"
"1. Should this be escalated to engineering? (yes/no)\n"
"2. Why or why not? (one sentence)\n"
"3. If yes, write an internal note for the engineering team - "
"include the customer impact, urgency, and what to investigate.\n"
"4. If no, confirm the support team can handle it and why.\n\n"
"Be direct. Engineers are busy - give them only what they need."
),
)
pipeline = classifier >> responder >> escalation_checker
ticket = """
Subject: URGENT - API extremely slow, blocking our monthly batch processing
From: David Park <david.park@acmelogistics.com>
Company: Acme Logistics (Enterprise plan)
Environment: Production (US-East)
Submitted: Monday 3:15 PM EST
Hi Support Team,
We're in the middle of our monthly batch processing and we're completely
stuck. This is our most critical operational window of the month.
Here's what's happening:
1. Our batch API calls are going through, but response times have gone
from ~200ms to 15–30 SECONDS per call. We have 50,000+ items to
process and at this rate it will take days instead of hours.
2. The web dashboard is nearly unusable - pages take 45+ seconds to
load, and we keep getting timeout errors when trying to view our
job status.
3. We tried splitting our batch into smaller chunks thinking it was a
rate limit issue, but even individual API calls are slow.
4. Nothing changed on our end - same code, same volume as last month
when everything ran fine in under 2 hours.
We have downstream systems waiting on this data and our SLA with our
own customers is at risk. Three of our team members have been stuck
on this since 1 PM and can't do anything else until it's resolved.
Can someone please look into this ASAP? We need to know:
- Is there a known issue on your end?
- Is there anything we can do to work around it?
- When can we expect normal performance?
Thanks,
David Park
Senior Platform Engineer, Acme Logistics
"""

with AgentRuntime() as runtime:
result = runtime.run(pipeline, ticket)
result.print_result()

Going further: connect to real Zendesk

The hardcoded ticket was fine for learning >>. But in production, you would want the pipeline to read a ticket from Zendesk, post the response as a comment, and add the escalation note end-to-end. Here’s how.

Store your credentials

In the AgentSpan dashboard (localhost:6767), go to Credentials and add:

Name Value
──────────────────────────────────────────
ZENDESK_EMAIL Your Zendesk email
ZENDESK_API Your Zendesk API token

The tools

Three tools, one for each stage of the pipeline. Each tool declares which credentials it needs with credentials=[…]. The server fetches them from the credential store and injects them into os.environ when the tool runs:

from agentspan.agents import tool
import os
import requests

ZENDESK_SUBDOMAIN = "yourcompany"
@tool(credentials=["ZENDESK_EMAIL", "ZENDESK_API"])
def get_ticket(ticket_id: int) -> dict:
"""Fetch a support ticket from Zendesk by ID."""
auth = (f"{os.environ['ZENDESK_EMAIL']}/token", os.environ["ZENDESK_API"])
resp = requests.get(
f"https://{ZENDESK_SUBDOMAIN}.zendesk.com/api/v2/tickets/{ticket_id}.json",
auth=auth,
)
ticket = resp.json()["ticket"]
return {
"id": ticket["id"],
"subject": ticket["subject"],
"description": ticket["description"],
"status": ticket["status"],
"priority": ticket["priority"],
"created_at": ticket["created_at"],
}
@tool(credentials=["ZENDESK_EMAIL", "ZENDESK_API"])
def reply_to_ticket(ticket_id: int, message: str) -> dict:
"""Post a public comment on a Zendesk ticket. The customer will see this."""
auth = (f"{os.environ['ZENDESK_EMAIL']}/token", os.environ["ZENDESK_API"])
resp = requests.put(
f"https://{ZENDESK_SUBDOMAIN}.zendesk.com/api/v2/tickets/{ticket_id}.json",
auth=auth,
json={"ticket": {"comment": {"body": message, "public": True}}},
)
return {"status": "posted", "ticket_id": ticket_id}
@tool(credentials=["ZENDESK_EMAIL", "ZENDESK_API"])
def add_internal_note(ticket_id: int, note: str) -> dict:
"""Add a private internal note on a Zendesk ticket. Only your team sees this."""
auth = (f"{os.environ['ZENDESK_EMAIL']}/token", os.environ["ZENDESK_API"])
resp = requests.put(
f"https://{ZENDESK_SUBDOMAIN}.zendesk.com/api/v2/tickets/{ticket_id}.json",
auth=auth,
json={"ticket": {"comment": {"body": note, "public": False}}},
)
return {"status": "noted", "ticket_id": ticket_id}

Wire them in

Each agent gets the tools it needs:

classifier = Agent(
name="classifier",
model="openai/gpt-4o",
instructions="…", # same as before
tools=[get_ticket],
)

responder = Agent(
name="responder",
model="openai/gpt-4o",
instructions=(
"You are a senior support engineer. Given the classification, "
"draft a reply and post it as a comment on the ticket using "
"the reply_to_ticket tool."
),
tools=[reply_to_ticket],
)

escalation_checker = Agent(
name="escalation_checker",
model="openai/gpt-4o",
instructions=(
"You are a support team lead. Decide if this needs engineering "
"escalation. If yes, add an internal note on the ticket using "
"the add_internal_note tool."
),
tools=[add_internal_note],
)

pipeline = classifier >> responder >> escalation_checker
with AgentRuntime() as runtime:
result = runtime.run(pipeline, "Triage Zendesk ticket #7221")
result.print_result()

The >> pipeline is identical. The only change is that each agent now has tools to read from and write to Zendesk. The classifier fetches the ticket. The responder posts a public comment — the customer sees it in their ticket thread. The escalation checker adds a private internal note — only your team sees it.

That’s the full loop: ticket comes in, pipeline runs, customer gets a response, engineering gets the escalation note. No human had to copy-paste anything.

Why sequential?

The sequential strategy is the right choice when each agent’s work depends on the previous agent’s output. The responder needs the classification to write a relevant reply. The escalation checker needs both the classification and draft response to make the right call.

But not every problem is sequential. What if you wanted to run the classifier and a sentiment analyzer simultaneously? That’s the parallel multi-agent strategy, and it’s the topic of the next post in this series.

Composability

Want to extend the pipeline? Just add more agents to the chain:

Add a Jira ticket creator

pipeline = classifier >> responder >> escalation_checker >> jira_agent

Or add a translator for global support teams

pipeline = classifier >> responder >> escalation_checker >> translator

Same Agent class, same >> operator. The pipeline grows linearly — no new abstractions, no framework boilerplate.

And when you’re ready to go beyond sequential, AgentSpan has eight multi-agent strategies, all using the same Agent class:

Try it

pip install agentspan
agentspan server start

GitHub : github.com/agentspan/agentspan
Blog examples: github.com/agentspan/agentspan/tree/main/sdk/python/examples/blog_and_videos/sequential_pipeline
Docs : agentspan.ai/docs
Discord: https://discord.com/invite/ajcA66JcKq

What’s next

Part 2: The Parallel Strategy — Multiple agents tackle the same input simultaneously, and their results arrive together. Same Agent class, different strategy.