Routing OpenAI’s Swarm Framework & Multi-Agent Handoffs to Human Sales Teams in Real Time
Executive Summary
Multi-agent AI systems are increasingly used for lead qualification and top-of-funnel discovery. OpenAI’s Swarm was an experimental, educational Python framework built around two primitives, agents and handoffs. OpenAI’s repository now states that Swarm has been replaced by the OpenAI Agents SDK, which it describes as a production-ready evolution of Swarm, and recommends migrating for all production use cases. The Agents SDK is available for Python and for JavaScript/TypeScript.
In a typical AI-assisted revenue pipeline, specialized agents split the work:
- A prospecting agent verifies firmographic data.
- A discovery (BDR-style) agent holds the first conversations.
- A technical scoping agent gathers requirements and checks fit.
Qualification can be automated, but the close usually needs a person for negotiation, relationship building and contract execution. The hard engineering problem is therefore the human interception: reliably catching the moment agents finish their work and routing the deal, with its context, to a human Account Executive (AE) fast enough to matter.
Speed matters because response time affects outcomes. A widely cited 2011 Harvard Business Review study, The Short Life of Online Sales Leads, audited 2,241 U.S. companies by submitting test web leads. Only 37% responded within an hour, 23% never responded, and the average response among responders took about 42 hours. Firms that contacted a lead within an hour were nearly seven times as likely to qualify it as firms that waited even one hour longer. The study is more than a decade old and measured web-form leads, so treat it as directional rather than as a benchmark for AI-routed handoffs.
This guide covers how to intercept agent handoffs and route them to humans through Slack and Microsoft Teams, CRMs such as Salesforce and HubSpot, and telephony via Twilio. It uses Swarm to illustrate the concept and the Agents SDK for the production-oriented pattern.
Swarm vs. the OpenAI Agents SDK
| | Swarm | OpenAI Agents SDK |
| --- | --- | --- |
| Status | Experimental, educational; replaced by the Agents SDK | Actively maintained by OpenAI; positioned for production |
| Core primitives | Agents and handoffs | Agents, handoffs, guardrails, sessions, tracing |
| Underlying API | Chat Completions API | Built on the Responses API |
| Memory | Stateless; you pass the history in on every call | Optional session objects for conversation memory |
| Languages | Python 3.10+ | Python and JavaScript/TypeScript |
| Observability | None built in | Built-in tracing, with support for custom trace processors |
| Human-in-the-loop | Not built in | Tool-approval flow that pauses a run and resumes from a serializable RunState |
If you are starting a new project, build on the Agents SDK. The Swarm examples below remain useful because they show handoff mechanics in very little code, and the interception patterns carry over.
Understanding the Handoff Architecture
Agents, Routines and Handoffs
Swarm’s README describes two primitives: agents and handoffs.
1. Agents bundle instructions with a set of tools (plain Python functions).
2. Handoffs occur when a function returns another Agent. The run loop then switches to that agent’s instructions and tools while keeping the conversation history.
“Routines” (a set of instructions plus the tools needed to carry them out) is a pattern described in OpenAI’s cookbook material that Swarm was designed to demonstrate. It is a design idea rather than a separate class in the library.
In the Agents SDK, a handoff is also modeled as a tool call. The model sees each handoff target as a tool named transfer_to_<agent_name> by default. By default the receiving agent sees the entire prior conversation. You can change this with an input_filter, run custom logic with an on_handoff callback, and require structured input from the model via an input schema. The SDK also ships a recommended prompt prefix (RECOMMENDED_PROMPT_PREFIX) to help models use handoffs reliably.
+-------------------------------------------------------------------------+
| Autonomous Sub-Agent Loop |
| |
| +--------------------+ Handoff +-----------------------------+ |
| | Discovery Agent | -------------> | Technical Scoping Agent | |
| +--------------------+ +-----------------------------+ |
+-------------------------------------------------------|-----------------+
|
Escalation tool call | (halts the loop)
v
+-------------------------------------------------------------------------+
| Human AE Escalation & Routing Layer |
| |
| +-----------------------+ +-------------------+ +-----------------+ |
| | Real-time alert | | CRM update | | Voice / chat | |
| | (Slack / Teams) | | (Salesforce/HubS) | | takeover | |
| +-----------------------+ +-------------------+ +-----------------+ |
+-------------------------------------------------------------------------+
Context and State
- Swarm passes the message history plus a context_variables dictionary. If a tool function declares a context_variables parameter, Swarm injects it automatically and hides it from the model-facing schema. Swarm is stateless between calls, so your application must persist the history and context.
- Agents SDK passes a typed context object to tools through RunContextWrapper. The context is local to your code and is not sent to the model. Sessions can persist conversation history between runs.
Why Human Handoffs Are Different
A normal handoff assumes another agent is ready to take the next turn. A human AE introduces three frictions:
1. Asynchronous availability. An AE’s availability depends on calendars, routing rules and presence, while the agent loop expects an immediate tool result.
2. Context fragmentation. A human cannot read 40 turns of raw transcript in real time. They need a structured brief.
3. Ownership and state. Once the handoff fires, the AI must stop replying to the prospect. Otherwise the AI and the human can answer the same message.
Pattern 1: Intercepting a Swarm Handoff
In Swarm, the simplest interception is a special tool whose job is to stop the autonomous loop and pass control to your routing layer. The example below raises a custom exception. Swarm’s run loop calls your tool functions directly, so an exception raised inside a tool propagates out of client.run() unless you catch it, and you handle it in the calling code.
Two caveats apply to this pattern. First, the exception approach is a pragmatic hack suited to prototypes. Second, any changes made to context_variables inside the tool are lost when the exception propagates, so the payload must be built in the tool and carried on the exception.
import json
import logging
from typing import Any, Dict, List
from swarm import Swarm, Agent
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("SalesHandoffLogger")
client = Swarm()
class HumanHandoffException(Exception):
"""Raised by the escalation tool to break out of the autonomous loop."""
def __init__(self, agent_name: str, payload: Dict[str, Any]):
self.agent_name = agent_name
self.payload = payload
super().__init__(f"Handoff intercepted from {agent_name} to a human AE.")
def transfer_to_technical_scoping() -> Agent:
"""Hand the conversation to the technical scoping agent."""
return tech_scoping_agent
def escalate_to_human_ae(
context_variables: Dict[str, Any],
qualification_summary: str,
urgency_level: str = "HIGH",
) -> Agent:
"""Call when qualification criteria are met and a human Account
Executive is required to continue."""
payload = {
"session_id": context_variables.get("session_id"),
"lead_profile": {
"crm_lead_id": context_variables.get("crm_lead_id"),
"email": context_variables.get("email"),
"company_name": context_variables.get("company_name"),
},
"qualification_data": {
"bant_score": context_variables.get("bant_score"),
"estimated_contract_value": context_variables.get("estimated_arr"),
"urgency": urgency_level,
"technical_requirements": context_variables.get("tech_stack", []),
},
"conversation_digest": {"summary": qualification_summary},
"status": "AWAITING_HUMAN",
}
raise HumanHandoffException("Technical Scoping Agent", payload)
tech_scoping_agent = Agent(
name="Technical Scoping Agent",
instructions=(
"You evaluate technical and security readiness. When the prospect has "
"confirmed their requirements and wants a custom quote, contract or demo, "
"call escalate_to_human_ae with a concise summary."
),
functions=[escalate_to_human_ae],
)
discovery_agent = Agent(
name="Discovery Agent",
instructions=(
"You are a discovery agent. Ask about team size, tech stack, budget and "
"timeline. Once budget is confirmed, call transfer_to_technical_scoping."
),
functions=[transfer_to_technical_scoping],
)
def run_sales_pipeline(messages: List[Dict[str, str]], context: Dict[str, Any]) -> Dict[str, Any]:
# Swarm is stateless: load the full history yourself and pass it in.
try:
response = client.run(
agent=discovery_agent,
messages=messages,
context_variables=context,
)
return {"status": "SUCCESS", "response": response.messages[-1]["content"]}
except HumanHandoffException as event:
logger.warning("Intercepted handoff: %s", json.dumps(event.payload))
dispatch_human_routing_pipeline(event.payload)
return {
"status": "HANDOFF_INTERCEPTED",
"message": "Connecting you with an Account Executive.",
"payload": event.payload,
}
def dispatch_human_routing_pipeline(payload: Dict[str, Any]) -> None:
# Publish to your event bus / webhook layer (see the routing section below).
logger.info("Dispatching handoff payload: %s", json.dumps(payload))
One limitation of this design is that the escalation message returned to the prospect is hardcoded in your application, and the run ends without a final model turn. That is usually what you want. Do not promise a specific wait time unless your routing layer can guarantee it.
Pattern 2: The Production Approach with the Agents SDK
The Agents SDK gives you cleaner ways to stop an agent loop without exceptions.
Halting the Loop with a Stop-at-Tool Behavior
An agent can be configured to end the run as soon as a specific tool is called, using tool_use_behavior. Because the run then ends with the tool’s output as the final result, your application sees a deterministic signal.
from dataclasses import dataclass, field
from typing import List
from agents import Agent, Runner, RunContextWrapper, StopAtTools, function_tool
@dataclass
class DealContext:
session_id: str
crm_lead_id: str
company_name: str
estimated_arr: int
tech_stack: List[str] = field(default_factory=list)
@function_tool
def escalate_to_human_ae(
ctx: RunContextWrapper[DealContext],
summary: str,
urgency: str = "HIGH",
) -> str:
"""Escalate the conversation to a human Account Executive."""
deal = ctx.context
mark_session_pending(deal.session_id) # your Redis helper (see below)
dispatch_human_routing_pipeline({ # your event-bus publisher
"session_id": deal.session_id,
"lead_profile": {
"crm_lead_id": deal.crm_lead_id,
"company_name": deal.company_name,
},
"qualification_data": {"estimated_contract_value": deal.estimated_arr},
"conversation_digest": {"summary": summary},
"urgency": urgency,
})
return "HANDOFF_REQUESTED"
scoping_agent = Agent[DealContext](
name="Technical Scoping Agent",
instructions=(
"Confirm technical and security requirements. When the prospect is ready "
"for a quote, contract or demo, call escalate_to_human_ae."
),
tools=[escalate_to_human_ae],
tool_use_behavior=StopAtTools(stop_at_tool_names=["escalate_to_human_ae"]),
)
discovery_agent = Agent[DealContext](
name="Discovery Agent",
instructions=(
"Qualify budget, authority, need and timeline. When budget is confirmed, "
"hand off to the Technical Scoping Agent."
),
handoffs=[scoping_agent],
)
# result = await Runner.run(discovery_agent, user_message, context=deal_context)
# if result.final_output == "HANDOFF_REQUESTED": the human path is already in motion.
Check the current Agents SDK documentation for exact signatures before shipping, as the API has evolved quickly.
Using the SDK’s Human-in-the-Loop Flow
The Agents SDK also has a built-in approval flow. A tool can declare needs_approval (either True or a function that decides per call). When the model calls such a tool, the run pauses and returns pending approvals as interruptions. You can serialize the run to a RunState, store it, and resume later after a person approves or rejects the call. The documentation says this flow is designed to be interruptible for long periods without keeping a server running.
This suits a different problem than lead routing. Use it when a human should approve a sensitive action by the agent, such as sending a contract or issuing a discount. Use the escalation-tool pattern above when the human should take over the conversation.
Voice Agents
For phone and voice channels, the Agents SDK also provides realtime agents (for example, RealtimeAgent and RealtimeSession in the JavaScript package). The docs describe an approval flow for realtime and voice sessions as well.
Real-Time Context Synthesis: The Sales Handoff Brief
Before you alert a rep, condense the conversation into a brief. A summarizer step (a separate LLM call, or a dedicated final agent) can map the transcript into a BANT (Budget, Authority, Need, Timeline) or MEDDPICC structure.
+-----------------------------------+
| Multi-Agent Execution History |
| (dozens of raw turns) |
+-----------------------------------+
|
v
+-----------------------------------+
| Context Transformation Layer |
| (LLM summarizer / structurer) |
+-----------------------------------+
|
v
+-----------------------------------------------------------+
| Structured Sales Brief (example) |
| |
| * Qualification score: 94/100 (internal model score) |
| * Budget: $75k-$100k ARR (stated by prospect) |
| * Pain points: legacy infrastructure latency, SOC 2 need |
| * Decision maker: VP Engineering (in conversation) |
| * Suggested next step: enterprise SLA, security review |
+-----------------------------------------------------------+
The score, figures and names in these examples are illustrative. Define your own scoring rubric, and have the summarizer mark which facts the prospect stated versus which the model inferred, so a rep knows what to verify on the first call.
JSON Schema for the Handoff Payload
A strict schema lets Slack, CRM and telephony consumers share one contract:
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "SwarmHumanSalesHandoff",
"type": "object",
"properties": {
"session_id": { "type": "string", "format": "uuid" },
"timestamp": { "type": "string", "format": "date-time" },
"lead_profile": {
"type": "object",
"properties": {
"crm_lead_id": { "type": "string" },
"first_name": { "type": "string" },
"last_name": { "type": "string" },
"email": { "type": "string", "format": "email" },
"company_name": { "type": "string" },
"employee_count": { "type": "integer" }
},
"required": ["crm_lead_id", "email", "company_name"]
},
"qualification_data": {
"type": "object",
"properties": {
"bant_score": { "type": "number", "minimum": 0, "maximum": 100 },
"budget_confirmed": { "type": "boolean" },
"estimated_contract_value": { "type": "number" },
"buying_timeline": { "type": "string" },
"identified_pain_points": { "type": "array", "items": { "type": "string" } }
},
"required": ["bant_score", "estimated_contract_value"]
},
"conversation_digest": {
"type": "object",
"properties": {
"summary": { "type": "string" },
"key_objections_raised": { "type": "array", "items": { "type": "string" } },
"last_prospect_message": { "type": "string" }
}
}
},
"required": ["session_id", "timestamp", "lead_profile", "qualification_data", "conversation_digest"]
}
Validate every payload against the schema before dispatch (for example, with the jsonschema library in Python). If validation fails, use the fallback described in the production checklist.
Routing Blueprints
Once a handoff is intercepted, publish it once to an event bus (Redis Pub/Sub, Kafka, SNS/SQS or similar) and let independent consumers handle each channel. A failure in one consumer, such as a CRM outage, should not block the Slack alert.
+-----------------------------------+
| Intercepted Handoff Payload |
+-----------------------------------+
|
v
+-----------------------------------+
| Event Bus (Redis Pub/Sub, etc.) |
+-----------------------------------+
|
+--------------------------+--------------------------+
| | |
v v v
+----------------+ +----------------+ +------------------+
| Slack / Teams | | CRM sync | | Voice / chat |
| alert consumer | | consumer | | takeover bridge |
+----------------+ +----------------+ +------------------+
Route 1: Slack and Microsoft Teams
Slack. Use Block Kit to send an actionable message with a claim button. Keep in mind:
- Slack expects an acknowledgement of interactive payloads (button clicks) within three seconds. Acknowledge immediately and do the claim logic asynchronously.
- Header blocks accept plain text only and are limited to 150 characters.
{
"blocks": [
{
"type": "header",
"text": { "type": "plain_text", "text": "AI handoff: high-intent lead", "emoji": true }
},
{
"type": "section",
"fields": [
{ "type": "mrkdwn", "text": "*Company:*\nAcme Corp (~$75k est. ARR)" },
{ "type": "mrkdwn", "text": "*Lead:*\nJane Doe (VP Engineering)" },
{ "type": "mrkdwn", "text": "*Qualification score:*\n94/100 (internal)" },
{ "type": "mrkdwn", "text": "*Urgency:*\nContract requested" }
]
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*AI summary:*\nProspect completed technical scoping and confirmed budget. Requests a custom security addendum before end of day."
}
},
{
"type": "actions",
"block_id": "handoff_actions",
"elements": [
{
"type": "button",
"text": { "type": "plain_text", "text": "Claim & open live chat" },
"style": "primary",
"value": "claim_session_LEAD-99823",
"action_id": "claim_lead_action"
},
{
"type": "button",
"text": { "type": "plain_text", "text": "Schedule AE demo" },
"value": "schedule_demo_LEAD-99823",
"action_id": "schedule_demo_action"
}
]
}
]
}
Microsoft Teams. Microsoft has been retiring Office 365 Connectors (the old incoming webhooks) in Teams in favor of the Workflows app (Power Automate). According to Microsoft’s Message Center notice MC1181996, the final deprecation rollout was scheduled for May 18–22, 2026, and connector-based webhooks had to be migrated to Workflows before then. Workflows webhooks accept Adaptive Cards. Interactive elements such as buttons and inputs are not supported for legacy MessageCard payloads, so build your alert as an Adaptive Card. Custom bot names and icons are not supported for Adaptive Card payloads posted this way. If you need richer interactivity, such as claim buttons that call your service, look into a Teams bot rather than a simple webhook.
Route 2: CRM Ingestion
Salesforce. Using the simple-salesforce library, update the lead and create a high-priority task for the owner. Two details commonly trip people up:
- Lead.AnnualRevenue is the prospect company’s annual revenue, not the expected contract value. Store deal size in a custom field (for example, Estimated_Contract_Value__c) or on an Opportunity after conversion.
- Status must be an existing value in your org’s Lead Status picklist. Add a value for AI-qualified leads, or use an existing one, rather than writing arbitrary text.
import logging
import os
from simple_salesforce import Salesforce
logger = logging.getLogger("SalesHandoffLogger")
def sync_handoff_to_salesforce(payload: dict) -> None:
sf = Salesforce(
username=os.environ["SF_USERNAME"],
password=os.environ["SF_PASSWORD"],
security_token=os.environ["SF_TOKEN"],
domain="login",
)
lead_id = payload["lead_profile"]["crm_lead_id"]
qual = payload["qualification_data"]
summary = payload["conversation_digest"]["summary"]
sf.Lead.update(lead_id, {
"Status": "AI Qualified - Awaiting AE", # must exist in your picklist
"Rating": "Hot", # standard Lead field
"Estimated_Contract_Value__c": qual["estimated_contract_value"], # custom field
})
sf.Task.create({
"WhoId": lead_id,
"Subject": f"AI handoff: follow up now ({payload['lead_profile']['company_name']})",
"Priority": "High",
"Status": "Not Started",
"Description": f"Qualification score: {qual['bant_score']}\n\nSummary:\n{summary}",
})
logger.info("Synced handoff for %s to Salesforce.", lead_id)
For production, prefer OAuth (for example, a connected app with the JWT bearer flow) over storing a username, password and security token. You will also need to assign the task to the right owner via OwnerId, which comes from your routing rules.
HubSpot. HubSpot’s CRM API exposes tasks and deals as objects you can create and associate with contacts and companies, so the same pattern applies: update the contact or deal stage, then create a task assigned to the owning rep.
Route 3: Voice Handoffs with Twilio
For phone calls, Twilio’s <Dial> verb supports a call whisper (also called call screening). Your service provides a URL on the <Number> noun. Twilio fetches it after the AE answers, and the AE hears a message while the caller continues to hear ringing. Per Twilio’s documentation, this requires answerOnBridge to be set to true. The whisper TwiML can contain <Say>, <Play>, <Gather> and <Hangup>, so the AE can accept or reject by pressing a key. It cannot contain another <Dial>.
The Node.js example below makes three practical choices:
- It looks up the AI summary by session ID instead of putting it in a URL query string, which avoids length limits and keeps customer data out of URLs and logs.
- It uses absolute URLs, which Twilio needs for webhooks.
- It accepts form-encoded bodies, the format Twilio uses for its webhook requests.
const express = require('express');
const VoiceResponse = require('twilio').twiml.VoiceResponse;
const app = express();
app.use(express.urlencoded({ extended: false })); // Twilio webhooks
app.use(express.json());
const BASE_URL = process.env.PUBLIC_BASE_URL; // e.g. https://routing.example.com
const handoffs = new Map(); // replace with Redis in production
// Called by your Python engine to register a handoff before redirecting the live call
app.post('/api/handoffs', (req, res) => {
const { sessionId, aePhone, aiSummary } = req.body;
handoffs.set(sessionId, { aePhone, aiSummary });
res.sendStatus(204);
});
// TwiML for the live call. Redirect the call here with Twilio's REST API
// (calls(callSid).update({ url: ... })).
app.post('/api/voice-handoff', (req, res) => {
const handoff = handoffs.get(req.query.session);
const twiml = new VoiceResponse();
if (!handoff) {
twiml.say('Sorry, we could not reach an account executive. We will call you back shortly.');
twiml.hangup();
} else {
twiml.say('Thanks for those details. Connecting you with an account executive now.');
const dial = twiml.dial({
callerId: process.env.TWILIO_COMPANY_NUMBER,
answerOnBridge: true, // required for whisper
action: `${BASE_URL}/api/voice-handoff-complete`,
});
dial.number({ url: `${BASE_URL}/api/ae-whisper?session=${encodeURIComponent(req.query.session)}` }, handoff.aePhone);
}
res.type('text/xml').send(twiml.toString());
});
// Whisper: heard only by the AE after answering, before the call is bridged
app.post('/api/ae-whisper', (req, res) => {
const handoff = handoffs.get(req.query.session);
const twiml = new VoiceResponse();
twiml.say(`Incoming AI handoff. ${handoff ? handoff.aiSummary : 'Qualified lead.'}`);
res.type('text/xml').send(twiml.toString());
});
app.listen(3000, () => console.log('Voice routing bridge on port 3000'));
You should also validate Twilio’s request signature on every webhook, handle the case where the AE does not answer (the action URL receives the dial result), and consider a fallback queue.
State Management and Preventing Race Conditions
When a conversation moves from AI to human, two things can go wrong: two reps claim the same deal, or the AI keeps answering after a human has taken over. Treat session ownership as explicit state, and make the claim operation atomic.
[ Prospect sends message ]
|
v
+-----------------------------+
| Look up session ownership |
+-----------------------------+
|
+-----------------+-----------------+
| | |
AI_ACTIVE HUMAN_PENDING HUMAN_OWNED
| | |
v v v
Run the agent loop Queue message, Forward to the
alert the rep owning AE's console
Atomic Claims with Redis
Redis’s SET command with the NX option sets a key only if it does not already exist, which makes it a simple building block for “first claim wins.” The plain SET calls in many tutorials are not safe for this purpose, because two reps can overwrite each other.
import json
import redis
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
PENDING_TTL_SECONDS = 120 # example claim window
def mark_session_pending(session_id: str) -> None:
r.set(f"session:{session_id}:pending", "1", ex=PENDING_TTL_SECONDS)
def claim_session(session_id: str, ae_id: str) -> bool:
"""Atomically claim a session. Returns True only for the first caller."""
claimed = r.set(f"session:{session_id}:owner", ae_id, nx=True)
if claimed:
r.delete(f"session:{session_id}:pending")
return bool(claimed)
def handle_incoming_prospect_message(session_id: str, text: str, context: dict):
owner = r.get(f"session:{session_id}:owner")
if owner: # HUMAN_OWNED
r.publish(f"channel:ae:{owner}", json.dumps({"session_id": session_id, "text": text}))
return {"action": "FORWARDED_TO_HUMAN"}
if r.exists(f"session:{session_id}:pending"): # HUMAN_PENDING
r.rpush(f"queue:session:{session_id}", text)
return {"action": "QUEUED_FOR_HUMAN"}
# AI_ACTIVE: load stored history, append the new message, run the agents
return run_sales_pipeline([{"role": "user", "content": text}], context)
Design notes:
- Ownership lives in the owner key, so the claim itself is the lock. For multi-step transitions that must be all-or-nothing, use a Lua script or a transaction.
- The pending key has a TTL, so an unclaimed lead expires into your fallback logic instead of waiting indefinitely. Make the expiry trigger an action (for example, via a scheduler or keyspace notifications). A silent expiry only removes the state.
- Persist ownership changes to a durable store as well, because Redis alone may not meet your audit needs.
Production Readiness Checklist
1. Latency and Timeouts
The following are example targets to tune against your own data, not industry standards:
- Alert latency: aim for the Slack or Teams alert to go out within a second or two of the handoff event.
- Claim timeout: if no AE claims within a window you choose (for example, 120 seconds), re-route to a fallback queue or have the AI offer to schedule a call.
Whatever targets you pick, measure them. The speed-to-lead research cited above is about how quickly a human reaches the lead, so track time from handoff trigger to the AE’s first contact, not only the alert.
2. Observability and Tracing
- The Agents SDK includes built-in tracing for agent runs (handoffs, tool calls, guardrails) and supports custom trace processors. Swarm has no built-in observability, so add your own.
- Common additions are OpenTelemetry, LangSmith and Weights & Biases Weave.
- Record spans or events for handoff_trigger_time, routing_latency and human_claim_time, and log every agent-to-agent transfer.
3. Guardrails and Safety
- The Agents SDK supports input and output guardrails. Use them to catch PII leakage and policy violations before they reach the prospect or your CRM.
- Add a handoff hop counter so agents cannot bounce a conversation between themselves indefinitely.
- Tell prospects when they are talking to an AI and when a human joins. Depending on your jurisdiction, rules on disclosure, call recording and automated calling may apply (for example, consent requirements for recording and restrictions on automated outbound calls), so consult counsel.
- Treat the prospect’s messages as untrusted input. A prospect could try to prompt-inject the agent into escalating, discounting or leaking internal notes, so keep decisions about pricing and contract terms in deterministic code or with the human.
4. Graceful Failure Recovery
- Summarization failure: if the brief cannot be generated or fails schema validation, send the last several raw turns so the AE still has context.
- Prospect disconnects while pending: notify the rep, and write the AI transcript and brief to the CRM record so the follow-up has context.
- Dependency outages: because each consumer reads from the event bus independently, a CRM or Slack outage should only delay that channel, not drop the handoff. Add retries with backoff and a dead-letter queue.
Conclusion
Whether you start from Swarm’s educational patterns or the Agents SDK’s production features, the human handoff is a routing problem as much as an AI problem. Four practices cover most of it:
1. Treat the handoff as an explicit event. Halt the agent loop with a dedicated escalation tool, using StopAtTools in the Agents SDK or a controlled exception in Swarm.
2. Synthesize before routing. Give reps a structured brief that separates stated facts from model inferences, instead of a raw transcript.
3. Lock ownership atomically. Use explicit session states and an atomic claim so only one rep owns a conversation and the AI stops replying.
4. Define and measure fallbacks. Set a claim timeout and an alternate route, and track time-to-human end to end.
Build these behaviors with explicit tests, such as simulated double-claims, dropped connections and summarizer failures, before putting real leads through the pipeline.
Sources and Further Reading
- OpenAI Swarm repository (README notes it is replaced by the Agents SDK)
- OpenAI Agents SDK (Python) documentation
- Agents SDK (JavaScript/TypeScript): Handoffs
- Agents SDK (Python): Human-in-the-loop
- Twilio: What is Call Whisper?
- Microsoft Message Center MC1181996: Office 365 connectors retirement in Teams
- Oldroyd, McElheran & Elkington, “The Short Life of Online Sales Leads,” Harvard Business Review, March 2011
---
Originally published at instachime.com
Mentions
Routing OpenAI’s Swarm Framework & Multi-Agent Handoffs to Human Sales Teams in Real Time Executive Summary Multi-agent AI systems are increasingly used for lead qualification and top-of-funnel discovery. OpenAI’s Swarm was an experimental, educational Python framework built around two primitives, agents and handoffs. OpenAI’s repository now states that Swarm has been replaced by the OpenAI Agents SDK, which it describes as a production-ready evolution of Swarm, and recommends migrating for all production use cases. The Agents SDK is available for Python and for JavaScript/TypeScript. In a typical AI-assisted revenue pipeline, specialized agents split the work: - A prospecting agent verifies firmographic data. - A discovery (BDR-style) agent holds the first conversations. - A technical scoping agent gathers requirements and checks fit. Qualification can be automated, but the close usually needs a person for negotiation, relationship building and contract execution. The hard engineering problem is therefore the human interception: reliably catching the moment agents finish their work and routing the deal, with its context, to a human Account Executive (AE) fast enough to matter. Speed matters because response time affects outcomes. A widely cited 2011 Harvard Business Review study, The Short Life of Online Sales Leads, audited 2,241 U.S. companies by submitting test web leads. Only 37% responded within an hour, 23% never responded, and the average response among responders took about 42 hours. Firms that contacted a lead within an hour were nearly seven times as likely to qualify it as firms that waited even one hour longer. The study is more than a decade old and measured web-form leads, so treat it as directional rather than as a benchmark for AI-routed handoffs. This guide covers how to intercept agent handoffs and route them to humans through Slack and Microsoft Teams, CRMs such as Salesforce and HubSpot, and telephony via Twilio. It uses Swarm to illustrate the concept and the Agents SDK for the production-oriented pattern. Swarm vs. the OpenAI Agents SDK | | Swarm | OpenAI Agents SDK | | --- | --- | --- | | Status | Experimental, educational; replaced by the Agents SDK | Actively maintained by OpenAI; positioned for production | | Core primitives | Agents and handoffs | Agents, handoffs, guardrails, sessions, tracing | | Underlying API | Chat Completions API | Built on the Responses API | | Memory | Stateless; you pass the history in on every call | Optional session objects for conversation memory | | Languages | Python 3.10+ | Python and JavaScript/TypeScript | | Observability | None built in | Built-in tracing, with support for custom trace processors | | Human-in-the-loop | Not built in | Tool-approval flow that pauses a run and resumes from a serializable RunState | If you are starting a new project, build on the Agents SDK. The Swarm examples below remain useful because they show handoff mechanics in very little code, and the interception patterns carry over. Understanding the Handoff Architecture Agents, Routines and Handoffs Swarm’s README describes two primitives: agents and handoffs. 1. Agents bundle instructions with a set of tools (plain Python functions). 2. Handoffs occur when a function returns another Agent. The run loop then switches to that agent’s instructions and tools while keeping the conversation history. “Routines” (a set of instructions plus the tools needed to carry them out) is a pattern described in OpenAI’s cookbook material that Swarm was designed to demonstrate. It is a design idea rather than a separate class in the library. In the Agents SDK, a handoff is also modeled as a tool call. The model sees each handoff target as a tool named transfer_to_<agent_name> by default. By default the receiving agent sees the entire prior conversation. You can change this with an input_filter, run custom logic with an on_handoff callback, and require structured input from the model via an input schema. The SDK also ships a recommended prompt prefix (RECOMMENDED_PROMPT_PREFIX) to help models use handoffs reliably. +-------------------------------------------------------------------------+ | Autonomous Sub-Agent Loop | | | | +--------------------+ Handoff +-----------------------------+ | | | Discovery Agent | -------------> | Technical Scoping Agent | | | +--------------------+ +-----------------------------+ | +-------------------------------------------------------|-----------------+ | Escalation tool call | (halts the loop) v +-------------------------------------------------------------------------+ | Human AE Escalation & Routing Layer | | | | +-----------------------+ +-------------------+ +-----------------+ | | | Real-time alert | | CRM update | | Voice / chat | | | | (Slack / Teams) | | (Salesforce/HubS) | | takeover | | | +-----------------------+ +-------------------+ +-----------------+ | +-------------------------------------------------------------------------+ Context and State - Swarm passes the message history plus a context_variables dictionary. If a tool function declares a context_variables parameter, Swarm injects it automatically and hides it from the model-facing schema. Swarm is stateless between calls, so your application must persist the history and context. - Agents SDK passes a typed context object to tools through RunContextWrapper. The context is local to your code and is not sent to the model. Sessions can persist conversation history between runs. Why Human Handoffs Are Different A normal handoff assumes another agent is ready to take the next turn. A human AE introduces three frictions: 1. Asynchronous availability. An AE’s availability depends on calendars, routing rules and presence, while the agent loop expects an immediate tool result. 2. Context fragmentation. A human cannot read 40 turns of raw transcript in real time. They need a structured brief. 3. Ownership and state. Once the handoff fires, the AI must stop replying to the prospect. Otherwise the AI and the human can answer the same message. Pattern 1: Intercepting a Swarm Handoff In Swarm, the simplest interception is a special tool whose job is to stop the autonomous loop and pass control to your routing layer. The example below raises a custom exception. Swarm’s run loop calls your tool functions directly, so an exception raised inside a tool propagates out of client.run() unless you catch it, and you handle it in the calling code. Two caveats apply to this pattern. First, the exception approach is a pragmatic hack suited to prototypes. Second, any changes made to context_variables inside the tool are lost when the exception propagates, so the payload must be built in the tool and carried on the exception. import json import logging from typing import Any, Dict, List from swarm import Swarm, Agent logging.basicConfig(level=logging.INFO) logger = logging.getLogger("SalesHandoffLogger") client = Swarm() class HumanHandoffException(Exception): """Raised by the escalation tool to break out of the autonomous loop.""" def __init__(self, agent_name: str, payload: Dict[str, Any]): self.agent_name = agent_name self.payload = payload super().__init__(f"Handoff intercepted from {agent_name} to a human AE.") def transfer_to_technical_scoping() -> Agent: """Hand the conversation to the technical scoping agent.""" return tech_scoping_agent def escalate_to_human_ae( context_variables: Dict[str, Any], qualification_summary: str, urgency_level: str = "HIGH", ) -> Agent: """Call when qualification criteria are met and a human Account Executive is required to continue.""" payload = { "session_id": context_variables.get("session_id"), "lead_profile": { "crm_lead_id": context_variables.get("crm_lead_id"), "email": context_variables.get("email"), "company_name": context_variables.get("company_name"), }, "qualification_data": { "bant_score": context_variables.get("bant_score"), "estimated_contract_value": context_variables.get("estimated_arr"), "urgency": urgency_level, "technical_requirements": context_variables.get("tech_stack", []), }, "conversation_digest": {"summary": qualification_summary}, "status": "AWAITING_HUMAN", } raise HumanHandoffException("Technical Scoping Agent", payload) tech_scoping_agent = Agent( name="Technical Scoping Agent", instructions=( "You evaluate technical and security readiness. When the prospect has " "confirmed their requirements and wants a custom quote, contract or demo, " "call escalate_to_human_ae with a concise summary." ), functions=[escalate_to_human_ae], ) discovery_agent = Agent( name="Discovery Agent", instructions=( "You are a discovery agent. Ask about team size, tech stack, budget and " "timeline. Once budget is confirmed, call transfer_to_technical_scoping." ), functions=[transfer_to_technical_scoping], ) def run_sales_pipeline(messages: List[Dict[str, str]], context: Dict[str, Any]) -> Dict[str, Any]: # Swarm is stateless: load the full history yourself and pass it in. try: response = client.run( agent=discovery_agent, messages=messages, context_variables=context, ) return {"status": "SUCCESS", "response": response.messages[-1]["content"]} except HumanHandoffException as event: logger.warning("Intercepted handoff: %s", json.dumps(event.payload)) dispatch_human_routing_pipeline(event.payload) return { "status": "HANDOFF_INTERCEPTED", "message": "Connecting you with an Account Executive.", "payload": event.payload, } def dispatch_human_routing_pipeline(payload: Dict[str, Any]) -> None: # Publish to your event bus / webhook layer (see the routing section below). logger.info("Dispatching handoff payload: %s", json.dumps(payload)) One limitation of this design is that the escalation message returned to the prospect is hardcoded in your application, and the run ends without a final model turn. That is usually what you want. Do not promise a specific wait time unless your routing layer can guarantee it. Pattern 2: The Production Approach with the Agents SDK The Agents SDK gives you cleaner ways to stop an agent loop without exceptions. Halting the Loop with a Stop-at-Tool Behavior An agent can be configured to end the run as soon as a specific tool is called, using tool_use_behavior. Because the run then ends with the tool’s output as the final result, your application sees a deterministic signal. from dataclasses import dataclass, field from typing import List from agents import Agent, Runner, RunContextWrapper, StopAtTools, function_tool @dataclass class DealContext: session_id: str crm_lead_id: str company_name: str estimated_arr: int tech_stack: List[str] = field(default_factory=list) @function_tool def escalate_to_human_ae( ctx: RunContextWrapper[DealContext], summary: str, urgency: str = "HIGH", ) -> str: """Escalate the conversation to a human Account Executive.""" deal = ctx.context mark_session_pending(deal.session_id) # your Redis helper (see below) dispatch_human_routing_pipeline({ # your event-bus publisher "session_id": deal.session_id, "lead_profile": { "crm_lead_id": deal.crm_lead_id, "company_name": deal.company_name, }, "qualification_data": {"estimated_contract_value": deal.estimated_arr}, "conversation_digest": {"summary": summary}, "urgency": urgency, }) return "HANDOFF_REQUESTED" scoping_agent = Agent[DealContext]( name="Technical Scoping Agent", instructions=( "Confirm technical and security requirements. When the prospect is ready " "for a quote, contract or demo, call escalate_to_human_ae." ), tools=[escalate_to_human_ae], tool_use_behavior=StopAtTools(stop_at_tool_names=["escalate_to_human_ae"]), ) discovery_agent = Agent[DealContext]( name="Discovery Agent", instructions=( "Qualify budget, authority, need and timeline. When budget is confirmed, " "hand off to the Technical Scoping Agent." ), handoffs=[scoping_agent], ) # result = await Runner.run(discovery_agent, user_message, context=deal_context) # if result.final_output == "HANDOFF_REQUESTED": the human path is already in motion. Check the current Agents SDK documentation for exact signatures before shipping, as the API has evolved quickly. Using the SDK’s Human-in-the-Loop Flow The Agents SDK also has a built-in approval flow. A tool can declare needs_approval (either True or a function that decides per call). When the model calls such a tool, the run pauses and returns pending approvals as interruptions. You can serialize the run to a RunState, store it, and resume later after a person approves or rejects the call. The documentation says this flow is designed to be interruptible for long periods without keeping a server running. This suits a different problem than lead routing. Use it when a human should approve a sensitive action by the agent, such as sending a contract or issuing a discount. Use the escalation-tool pattern above when the human should take over the conversation. Voice Agents For phone and voice channels, the Agents SDK also provides realtime agents (for example, RealtimeAgent and RealtimeSession in the JavaScript package). The docs describe an approval flow for realtime and voice sessions as well. Real-Time Context Synthesis: The Sales Handoff Brief Before you alert a rep, condense the conversation into a brief. A summarizer step (a separate LLM call, or a dedicated final agent) can map the transcript into a BANT (Budget, Authority, Need, Timeline) or MEDDPICC structure. +-----------------------------------+ | Multi-Agent Execution History | | (dozens of raw turns) | +-----------------------------------+ | v +-----------------------------------+ | Context Transformation Layer | | (LLM summarizer / structurer) | +-----------------------------------+ | v +-----------------------------------------------------------+ | Structured Sales Brief (example) | | | | * Qualification score: 94/100 (internal model score) | | * Budget: $75k-$100k ARR (stated by prospect) | | * Pain points: legacy infrastructure latency, SOC 2 need | | * Decision maker: VP Engineering (in conversation) | | * Suggested next step: enterprise SLA, security review | +-----------------------------------------------------------+ The score, figures and names in these examples are illustrative. Define your own scoring rubric, and have the summarizer mark which facts the prospect stated versus which the model inferred, so a rep knows what to verify on the first call. JSON Schema for the Handoff Payload A strict schema lets Slack, CRM and telephony consumers share one contract: { "$schema": "https://json-schema.org/draft/2020-12/schema", "title": "SwarmHumanSalesHandoff", "type": "object", "properties": { "session_id": { "type": "string", "format": "uuid" }, "timestamp": { "type": "string", "format": "date-time" }, "lead_profile": { "type": "object", "properties": { "crm_lead_id": { "type": "string" }, "first_name": { "type": "string" }, "last_name": { "type": "string" }, "email": { "type": "string", "format": "email" }, "company_name": { "type": "string" }, "employee_count": { "type": "integer" } }, "required": ["crm_lead_id", "email", "company_name"] }, "qualification_data": { "type": "object", "properties": { "bant_score": { "type": "number", "minimum": 0, "maximum": 100 }, "budget_confirmed": { "type": "boolean" }, "estimated_contract_value": { "type": "number" }, "buying_timeline": { "type": "string" }, "identified_pain_points": { "type": "array", "items": { "type": "string" } } }, "required": ["bant_score", "estimated_contract_value"] }, "conversation_digest": { "type": "object", "properties": { "summary": { "type": "string" }, "key_objections_raised": { "type": "array", "items": { "type": "string" } }, "last_prospect_message": { "type": "string" } } } }, "required": ["session_id", "timestamp", "lead_profile", "qualification_data", "conversation_digest"] } Validate every payload against the schema before dispatch (for example, with the jsonschema library in Python). If validation fails, use the fallback described in the production checklist. Routing Blueprints Once a handoff is intercepted, publish it once to an event bus (Redis Pub/Sub, Kafka, SNS/SQS or similar) and let independent consumers handle each channel. A failure in one consumer, such as a CRM outage, should not block the Slack alert. +-----------------------------------+ | Intercepted Handoff Payload | +-----------------------------------+ | v +-----------------------------------+ | Event Bus (Redis Pub/Sub, etc.) | +-----------------------------------+ | +--------------------------+--------------------------+ | | | v v v +----------------+ +----------------+ +------------------+ | Slack / Teams | | CRM sync | | Voice / chat | | alert consumer | | consumer | | takeover bridge | +----------------+ +----------------+ +------------------+ Route 1: Slack and Microsoft Teams Slack. Use Block Kit to send an actionable message with a claim button. Keep in mind: - Slack expects an acknowledgement of interactive payloads (button clicks) within three seconds. Acknowledge immediately and do the claim logic asynchronously. - Header blocks accept plain text only and are limited to 150 characters. { "blocks": [ { "type": "header", "text": { "type": "plain_text", "text": "AI handoff: high-intent lead", "emoji": true } }, { "type": "section", "fields": [ { "type": "mrkdwn", "text": "*Company:*\nAcme Corp (~$75k est. ARR)" }, { "type": "mrkdwn", "text": "*Lead:*\nJane Doe (VP Engineering)" }, { "type": "mrkdwn", "text": "*Qualification score:*\n94/100 (internal)" }, { "type": "mrkdwn", "text": "*Urgency:*\nContract requested" } ] }, { "type": "section", "text": { "type": "mrkdwn", "text": "*AI summary:*\nProspect completed technical scoping and confirmed budget. Requests a custom security addendum before end of day." } }, { "type": "actions", "block_id": "handoff_actions", "elements": [ { "type": "button", "text": { "type": "plain_text", "text": "Claim & open live chat" }, "style": "primary", "value": "claim_session_LEAD-99823", "action_id": "claim_lead_action" }, { "type": "button", "text": { "type": "plain_text", "text": "Schedule AE demo" }, "value": "schedule_demo_LEAD-99823", "action_id": "schedule_demo_action" } ] } ] } Microsoft Teams. Microsoft has been retiring Office 365 Connectors (the old incoming webhooks) in Teams in favor of the Workflows app (Power Automate). According to Microsoft’s Message Center notice MC1181996, the final deprecation rollout was scheduled for May 18–22, 2026, and connector-based webhooks had to be migrated to Workflows before then. Workflows webhooks accept Adaptive Cards. Interactive elements such as buttons and inputs are not supported for legacy MessageCard payloads, so build your alert as an Adaptive Card. Custom bot names and icons are not supported for Adaptive Card payloads posted this way. If you need richer interactivity, such as claim buttons that call your service, look into a Teams bot rather than a simple webhook. Route 2: CRM Ingestion Salesforce. Using the simple-salesforce library, update the lead and create a high-priority task for the owner. Two details commonly trip people up: - Lead.AnnualRevenue is the prospect company’s annual revenue, not the expected contract value. Store deal size in a custom field (for example, Estimated_Contract_Value__c) or on an Opportunity after conversion. - Status must be an existing value in your org’s Lead Status picklist. Add a value for AI-qualified leads, or use an existing one, rather than writing arbitrary text. import logging import os from simple_salesforce import Salesforce logger = logging.getLogger("SalesHandoffLogger") def sync_handoff_to_salesforce(payload: dict) -> None: sf = Salesforce( username=os.environ["SF_USERNAME"], password=os.environ["SF_PASSWORD"], security_token=os.environ["SF_TOKEN"], domain="login", ) lead_id = payload["lead_profile"]["crm_lead_id"] qual = payload["qualification_data"] summary = payload["conversation_digest"]["summary"] sf.Lead.update(lead_id, { "Status": "AI Qualified - Awaiting AE", # must exist in your picklist "Rating": "Hot", # standard Lead field "Estimated_Contract_Value__c": qual["estimated_contract_value"], # custom field }) sf.Task.create({ "WhoId": lead_id, "Subject": f"AI handoff: follow up now ({payload['lead_profile']['company_name']})", "Priority": "High", "Status": "Not Started", "Description": f"Qualification score: {qual['bant_score']}\n\nSummary:\n{summary}", }) logger.info("Synced handoff for %s to Salesforce.", lead_id) For production, prefer OAuth (for example, a connected app with the JWT bearer flow) over storing a username, password and security token. You will also need to assign the task to the right owner via OwnerId, which comes from your routing rules. HubSpot. HubSpot’s CRM API exposes tasks and deals as objects you can create and associate with contacts and companies, so the same pattern applies: update the contact or deal stage, then create a task assigned to the owning rep. Route 3: Voice Handoffs with Twilio For phone calls, Twilio’s <Dial> verb supports a call whisper (also called call screening). Your service provides a URL on the <Number> noun. Twilio fetches it after the AE answers, and the AE hears a message while the caller continues to hear ringing. Per Twilio’s documentation, this requires answerOnBridge to be set to true. The whisper TwiML can contain <Say>, <Play>, <Gather> and <Hangup>, so the AE can accept or reject by pressing a key. It cannot contain another <Dial>. The Node.js example below makes three practical choices: - It looks up the AI summary by session ID instead of putting it in a URL query string, which avoids length limits and keeps customer data out of URLs and logs. - It uses absolute URLs, which Twilio needs for webhooks. - It accepts form-encoded bodies, the format Twilio uses for its webhook requests. const express = require('express'); const VoiceResponse = require('twilio').twiml.VoiceResponse; const app = express(); app.use(express.urlencoded({ extended: false })); // Twilio webhooks app.use(express.json()); const BASE_URL = process.env.PUBLIC_BASE_URL; // e.g. https://routing.example.com const handoffs = new Map(); // replace with Redis in production // Called by your Python engine to register a handoff before redirecting the live call app.post('/api/handoffs', (req, res) => { const { sessionId, aePhone, aiSummary } = req.body; handoffs.set(sessionId, { aePhone, aiSummary }); res.sendStatus(204); }); // TwiML for the live call. Redirect the call here with Twilio's REST API // (calls(callSid).update({ url: ... })). app.post('/api/voice-handoff', (req, res) => { const handoff = handoffs.get(req.query.session); const twiml = new VoiceResponse(); if (!handoff) { twiml.say('Sorry, we could not reach an account executive. We will call you back shortly.'); twiml.hangup(); } else { twiml.say('Thanks for those details. Connecting you with an account executive now.'); const dial = twiml.dial({ callerId: process.env.TWILIO_COMPANY_NUMBER, answerOnBridge: true, // required for whisper action: `${BASE_URL}/api/voice-handoff-complete`, }); dial.number({ url: `${BASE_URL}/api/ae-whisper?session=${encodeURIComponent(req.query.session)}` }, handoff.aePhone); } res.type('text/xml').send(twiml.toString()); }); // Whisper: heard only by the AE after answering, before the call is bridged app.post('/api/ae-whisper', (req, res) => { const handoff = handoffs.get(req.query.session); const twiml = new VoiceResponse(); twiml.say(`Incoming AI handoff. ${handoff ? handoff.aiSummary : 'Qualified lead.'}`); res.type('text/xml').send(twiml.toString()); }); app.listen(3000, () => console.log('Voice routing bridge on port 3000')); You should also validate Twilio’s request signature on every webhook, handle the case where the AE does not answer (the action URL receives the dial result), and consider a fallback queue. State Management and Preventing Race Conditions When a conversation moves from AI to human, two things can go wrong: two reps claim the same deal, or the AI keeps answering after a human has taken over. Treat session ownership as explicit state, and make the claim operation atomic. [ Prospect sends message ] | v +-----------------------------+ | Look up session ownership | +-----------------------------+ | +-----------------+-----------------+ | | | AI_ACTIVE HUMAN_PENDING HUMAN_OWNED | | | v v v Run the agent loop Queue message, Forward to the alert the rep owning AE's console Atomic Claims with Redis Redis’s SET command with the NX option sets a key only if it does not already exist, which makes it a simple building block for “first claim wins.” The plain SET calls in many tutorials are not safe for this purpose, because two reps can overwrite each other. import json import redis r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True) PENDING_TTL_SECONDS = 120 # example claim window def mark_session_pending(session_id: str) -> None: r.set(f"session:{session_id}:pending", "1", ex=PENDING_TTL_SECONDS) def claim_session(session_id: str, ae_id: str) -> bool: """Atomically claim a session. Returns True only for the first caller.""" claimed = r.set(f"session:{session_id}:owner", ae_id, nx=True) if claimed: r.delete(f"session:{session_id}:pending") return bool(claimed) def handle_incoming_prospect_message(session_id: str, text: str, context: dict): owner = r.get(f"session:{session_id}:owner") if owner: # HUMAN_OWNED r.publish(f"channel:ae:{owner}", json.dumps({"session_id": session_id, "text": text})) return {"action": "FORWARDED_TO_HUMAN"} if r.exists(f"session:{session_id}:pending"): # HUMAN_PENDING r.rpush(f"queue:session:{session_id}", text) return {"action": "QUEUED_FOR_HUMAN"} # AI_ACTIVE: load stored history, append the new message, run the agents return run_sales_pipeline([{"role": "user", "content": text}], context) Design notes: - Ownership lives in the owner key, so the claim itself is the lock. For multi-step transitions that must be all-or-nothing, use a Lua script or a transaction. - The pending key has a TTL, so an unclaimed lead expires into your fallback logic instead of waiting indefinitely. Make the expiry trigger an action (for example, via a scheduler or keyspace notifications). A silent expiry only removes the state. - Persist ownership changes to a durable store as well, because Redis alone may not meet your audit needs. Production Readiness Checklist 1. Latency and Timeouts The following are example targets to tune against your own data, not industry standards: - Alert latency: aim for the Slack or Teams alert to go out within a second or two of the handoff event. - Claim timeout: if no AE claims within a window you choose (for example, 120 seconds), re-route to a fallback queue or have the AI offer to schedule a call. Whatever targets you pick, measure them. The speed-to-lead research cited above is about how quickly a human reaches the lead, so track time from handoff trigger to the AE’s first contact, not only the alert. 2. Observability and Tracing - The Agents SDK includes built-in tracing for agent runs (handoffs, tool calls, guardrails) and supports custom trace processors. Swarm has no built-in observability, so add your own. - Common additions are OpenTelemetry, LangSmith and Weights & Biases Weave. - Record spans or events for handoff_trigger_time, routing_latency and human_claim_time, and log every agent-to-agent transfer. 3. Guardrails and Safety - The Agents SDK supports input and output guardrails. Use them to catch PII leakage and policy violations before they reach the prospect or your CRM. - Add a handoff hop counter so agents cannot bounce a conversation between themselves indefinitely. - Tell prospects when they are talking to an AI and when a human joins. Depending on your jurisdiction, rules on disclosure, call recording and automated calling may apply (for example, consent requirements for recording and restrictions on automated outbound calls), so consult counsel. - Treat the prospect’s messages as untrusted input. A prospect could try to prompt-inject the agent into escalating, discounting or leaking internal notes, so keep decisions about pricing and contract terms in deterministic code or with the human. 4. Graceful Failure Recovery - Summarization failure: if the brief cannot be generated or fails schema validation, send the last several raw turns so the AE still has context. - Prospect disconnects while pending: notify the rep, and write the AI transcript and brief to the CRM record so the follow-up has context. - Dependency outages: because each consumer reads from the event bus independently, a CRM or Slack outage should only delay that channel, not drop the handoff. Add retries with backoff and a dead-letter queue. Conclusion Whether you start from Swarm’s educational patterns or the Agents SDK’s production features, the human handoff is a routing problem as much as an AI problem. Four practices cover most of it: 1. Treat the handoff as an explicit event. Halt the agent loop with a dedicated escalation tool, using StopAtTools in the Agents SDK or a controlled exception in Swarm. 2. Synthesize before routing. Give reps a structured brief that separates stated facts from model inferences, instead of a raw transcript. 3. Lock ownership atomically. Use explicit session states and an atomic claim so only one rep owns a conversation and the AI stops replying. 4. Define and measure fallbacks. Set a claim timeout and an alternate route, and track time-to-human end to end. Build these behaviors with explicit tests, such as simulated double-claims, dropped connections and summarizer failures, before putting real leads through the pipeline. Sources and Further Reading - OpenAI Swarm repository (README notes it is replaced by the Agents SDK) - OpenAI Agents SDK (Python) documentation - Agents SDK (JavaScript/TypeScript): Handoffs - Agents SDK (Python): Human-in-the-loop - Twilio: What is Call Whisper? - Microsoft Message Center MC1181996: Office 365 connectors retirement in Teams - Oldroyd, McElheran & Elkington, “The Short Life of Online Sales Leads,” Harvard Business Review, March 2011 --- Originally published at instachime.com
10/ OpenAI Swarm — lightweight educational examples of agent handoffs. openai/swarm Treat it as an experimental learning resource, not a production-ready framework.
oh-my-mermaid × Codex で Anthropic・OpenAI の公式リポジトリ5つを構成図にしてみた はじめに oh-my-mermaid（omm）は、Claude Code や Codex などの AI コーディングツールにスキルを登録し、リポジトリを解析させて Mermaid の構成図と説明を .omm/ フォルダに残す CLI です。 Anthropic と OpenAI の公式リポジトリ5つを、同じ条件で omm＋Codex CLI に解析させ、できた図を並べてみました。 結論 - 5つとも1回で構成図と日本語の説明ができ、omm validate（図の文法・規約の検査）も全件通った - 1リポジトリあたり 68〜89秒、3.2万〜4.7万トークン（5つで約21.6万トークン） - 図を並べると、Claude Agent SDK は「Claude Code CLI をサブプロセスで動かす薄い層」、OpenAI Agents SDK は「Runner がモデル・ツール・履歴・トレースを束ねる自前の実行系」と、設計の違いがひと目でわかった - 費用を抑えるため「観点は3つまで・2段まで」に制限したので、15万行の SDK でも図は5〜7ノードの粗い要約になる 環境 - WSL2 の Ubuntu 22.04（隔離した試用環境） - oh-my-mermaid 0.2.0、Codex CLI 0.162.1（OpenAI の API キーで実行） 手順 前提: Codex CLI を OpenAI の API キーで動かすため、環境変数 CODEX_API_KEY に API キーを設定済みであること（キーの値はここには書きません）。解析ごとに API の料金がかかります。 npm install -g oh-my-mermaid@0.2.0 @openai/codex@0.162.1 omm setup codex # ~/.agents/skills/oh-my-mermaid/ にスキルを登録 git clone --depth 1 https://github.com/anthropics/claude-agent-sdk-python.git cd claude-agent-sdk-python omm config language ja # 説明を日本語にする codex exec --skip-git-repo-check --dangerously-bypass-approvals-and-sandbox \ "Use the omm-scan skill (~/.agents/skills/oh-my-mermaid/omm-scan/SKILL.md) to scan this repository (focus on the src/claude_agent_sdk package) and generate .omm/ documentation. To limit cost: choose at most 3 perspectives, recurse at most 2 levels below each perspective, and keep each field short. Do not run the project's code or install anything." \ < /dev/null > ../scan.log 2>&1 omm status omm validate - プロンプトの括弧内（解析の対象）だけをリポジトリごとに変えました - --dangerously-bypass-approvals-and-sandbox は承認なしでコマンドを実行させるオプションです。今回は使い捨ての隔離環境の中なので使いました。普段の PC では使わないでください 結果 | リポジトリ | 対象の規模（Python） | 秒 | トークン | 観点 | 図 | |---|---|---|---|---|---| | anthropics/claude-agent-sdk-python | 26ファイル・13,295行 | 86 | 47,259 | 2 | 6 | | anthropics/anthropic-sdk-python | 2,288ファイル・155,171行 | 85 | 43,285 | 2 | 7 | | anthropics/claude-code | 22ファイル・7,914行＋プラグイン等 | 89 | 46,626 | 2 | 8 | | openai/openai-agents-python | 320ファイル・135,502行 | 71 | 47,227 | 3 | 4 | | openai/swarm | 6ファイル・512行 | 68 | 31,587 | 2 | 3 | 観点（perspective）は AI が選びます。全件で全体構成（overall-architecture）が選ばれ、ほかにデータの流れ（data-flow）、リクエストの流れ（request-lifecycle）、拡張点（extension-points）が選ばれました。 規模が300倍違っても、時間・トークン数はほぼ同じでした。上限をかけたため、大きいリポジトリほど粗く要約されるということです。 Claude Agent SDK（Python） graph LR public-api["公開API\nsrc/claude_agent_sdk/"] runtime["非同期実行\nsrc/claude_agent_sdk/_internal/"] transport["通信\nsrc/claude_agent_sdk/_internal/transport/"] session-services["セッション管理\nsrc/claude_agent_sdk/_internal/session*.py"] sdk-tools["SDKツール\nsrc/claude_agent_sdk/__init__.py"] contracts["型・例外\nsrc/claude_agent_sdk/types.py"] public-api -->|"実行を委譲"| runtime runtime -->|"制御・入出力"| transport public-api -->|"履歴APIを公開"| session-services runtime -->|"再開・ミラー保存"| session-services sdk-tools -->|"MCPサーバーを提供"| runtime public-api -->|"型を公開"| contracts 通信層の実体は SubprocessCLITransport（Claude Code CLI をサブプロセスで起動）で、create_sdk_mcp_server で Python の関数を MCP ツールとして渡せる、という構造がソースと一致しました。 OpenAI Agents SDK（Python） graph LR agent-definition["Agent 定義\nsrc/agents/agent.py"] runtime["実行制御\nsrc/agents/run.py"] models["モデル接続\nsrc/agents/models/"] tools["ツールと MCP\nsrc/agents/tool.py"] state["履歴と再開\nsrc/agents/memory/"] tracing["トレース\nsrc/agents/tracing/"] media["音声経路\nsrc/agents/realtime/"] agent-definition -->|"設定を渡す"| runtime runtime -->|"応答を要求"| models runtime -->|"呼び出しを実行"| tools runtime -->|"履歴を読書き・中断を保存"| state runtime -->|"スパンを記録"| tracing media -->|"共通トレースを利用"| tracing Claude Agent SDK がエージェントのループを CLI 側に任せるのに対し、こちらは Runner（run.py）が自前でループを回し、履歴（SQLite などのセッション）やトレースまで持っています。 Anthropic SDK（Python） graph LR client["公開クライアント\nsrc/anthropic/_client.py"] resources["API リソース\nsrc/anthropic/resources/"] transport["共通通信基盤\nsrc/anthropic/_base_client.py"] schemas["型とモデル\nsrc/anthropic/types/"] helpers["高水準の補助機能\nsrc/anthropic/lib/"] client -->|"リソースを公開"| resources client -->|"継承して通信"| transport resources -->|"HTTP 操作を委譲"| transport resources -->|"入出力型を参照"| schemas transport -->|"応答モデルを構築"| schemas helpers -->|"クライアントを拡張"| client resources -->|"stream・parse を利用"| helpers 懸念（concern）の欄に「生成 API と手書き補助層が混在」と書かれました。15万行の大半は自動生成のコードで、図では resources/ の1ノードにまとまっています。 Claude Code（公開リポジトリ） graph LR plugins["公開プラグイン\nplugins/"] examples["導入例\nexamples/"] automation["運用自動化\nscripts/; .github/"] mods["組み込み mod\nmods/"] dev-environment["開発環境\n.devcontainer/; Script/"] claude-host["Claude Code ホスト\n外部配布 CLI"] plugins -->|"拡張を読み込む"| claude-host examples -->|"設定・導入のひな形"| claude-host mods -->|"組み込みフックを提供"| claude-host automation -->|"Action・mod 検証で利用"| claude-host dev-environment -->|"実行環境を用意"| claude-host 公開リポジトリには CLI 本体の実装が無く、プラグインや周辺資産が中心です。omm も説明に「CLI 本体の実装全体ではなく、周辺資産と mods のソースを収録」と正しく書きました。 Swarm graph LR public-api["公開 API\nswarm/__init__.py"] core["実行エンジン\nswarm/core.py"] models["データ型\nswarm/types.py"] utilities["補助関数\nswarm/util.py"] repl["対話ループ\nswarm/repl/repl.py"] public-api -->|"Swarm・Agent・Response を公開"| core public-api -->|"型を公開"| models core -->|"状態と戻り値を表現"| models core -->|"スキーマ化・差分結合・ログ"| utilities repl -->|"run を呼び出す"| public-api 制約の欄に「呼び出し間の状態は呼び出し側が保持。永続ストアや HTTP サーバーはない」とあり、教育用の軽量フレームワークという性格をよく表しています。 内容はソースと合っているか 全部は確認していませんが、各リポジトリで1〜2点を grep で抜き取り確認し、すべて一致しました。 | リポジトリ | 図の関係 | ソース | |---|---|---| | claude-agent-sdk-python | 通信は CLI のサブプロセス | _internal/transport/subprocess_cli.py の class SubprocessCLITransport(Transport) | | claude-agent-sdk-python | SDK が MCP サーバーを提供 | __init__.py の def create_sdk_mcp_server( | | anthropic-sdk-python | クライアントが通信基盤を継承 | _client.py の class Anthropic(SyncAPIClient) | | openai-agents-python | 実行制御がトレースを記録 | run.py の run_state.set_trace(get_current_trace()) | | swarm | 対話ループが run を呼ぶ | repl/repl.py の client.run( | 所感 - 知らない SDK を読み始める前の「入口の地図」として便利です。5つとも同じ形（観点・図・説明）で出るので、横に並べて比べやすいのが良い点でした - 説明は1〜2文と短めです。大きいリポジトリを深く知りたいときは、上限を緩めるか、サブフォルダごとに解析するのがよさそうです（未確認） - 結果はテキストなので git で管理でき、omm diff で前回の図との差分も見られます - 注意: ビューア（omm view）は試した版（0.2.0）では全インターフェースで待ち受けました。共有のネットワークで使うときは気を付けてください 参考 - oh-my-mermaid/oh-my-mermaid - anthropics/claude-agent-sdk-python - anthropics/anthropic-sdk-python - anthropics/claude-code - openai/openai-agents-python - openai/swarm