Lightweight framework for building multi-agent workflows with LLMs, supporting handoffs, guardrails, tools, and 100+ LLM providers
Handoffs enable agent-to-agent delegation in multi-agent workflows. They are implemented as specialized tool calls that transfer control from one agent to another, with support for input filtering, history management, and custom handoff logic.
Configuration for agent handoffs with customizable behavior.
class Handoff[TContext, TAgent]:
"""
Agent handoff configuration.
Type Parameters:
- TContext: Type of context object
- TAgent: Type of target agent
Attributes:
- tool_name: str - Tool name for handoff
- tool_description: str - Tool description for LLM
- input_json_schema: dict[str, Any] - Input schema
- on_invoke_handoff: Callable - Invocation function
- agent_name: str - Target agent name
- input_filter: HandoffInputFilter | None - Input filter function
- nest_handoff_history: bool | None - History nesting override
- strict_json_schema: bool - Use strict JSON schema mode
- is_enabled: bool | Callable - Enabled state or function
"""
def get_transfer_message(agent: Agent) -> str:
"""
Get transfer message for handoff.
Parameters:
- agent: Target agent
Returns:
- str: Transfer message
"""
@classmethod
def default_tool_name(agent: Agent) -> str:
"""
Get default tool name for agent.
Parameters:
- agent: Agent to generate name for
Returns:
- str: Default tool name
"""
@classmethod
def default_tool_description(agent: Agent) -> str:
"""
Get default description for agent.
Parameters:
- agent: Agent to generate description for
Returns:
- str: Default description
"""Helper function to create handoffs from agents.
def handoff(
agent: Agent,
*,
tool_name_override: str | None = None,
tool_description_override: str | None = None,
on_handoff: Callable | None = None,
input_type: type | None = None,
input_filter: HandoffInputFilter | None = None,
nest_handoff_history: bool | None = None,
is_enabled: bool | Callable = True
) -> Handoff[TContext, Agent[TContext]]:
"""
Create handoff from agent.
Parameters:
- agent: Target agent for handoff
- tool_name_override: Custom tool name (default: "transfer_to_{agent_name}")
- tool_description_override: Custom description
- on_handoff: Callback when handoff occurs
- input_type: Type for handoff input parameters
- input_filter: Function to filter/transform handoff input
- nest_handoff_history: Whether to nest history in single message
- is_enabled: Whether handoff is enabled or function to determine
Returns:
- Handoff: Configured handoff object
"""Usage example:
from agents import Agent, handoff
support_agent = Agent(
name="Support Agent",
instructions="Handle customer support requests."
)
sales_agent = Agent(
name="Sales Agent",
instructions="Handle sales inquiries."
)
# Simple handoff
triage_agent = Agent(
name="Triage Agent",
instructions="Route to appropriate agent.",
handoffs=[support_agent, sales_agent]
)
# Custom handoff with callback
def on_transfer_to_sales(ctx, agent, input_data):
print(f"Transferring to sales with: {input_data}")
custom_handoff = handoff(
sales_agent,
tool_name_override="escalate_to_sales",
tool_description_override="Escalate to sales team for complex inquiries",
on_handoff=on_transfer_to_sales
)
triage_agent = Agent(
name="Triage Agent",
handoffs=[support_agent, custom_handoff]
)Data structure containing handoff context and history.
class HandoffInputData:
"""
Input data for handoffs.
Attributes:
- input_history: str | tuple[TResponseInputItem, ...] - Input history
- pre_handoff_items: tuple[RunItem, ...] - Items before handoff
- new_items: tuple[RunItem, ...] - New items including handoff
- run_context: RunContextWrapper | None - Run context
"""
def clone(**kwargs) -> HandoffInputData:
"""
Create modified copy.
Parameters:
- **kwargs: Fields to override
Returns:
- HandoffInputData: New instance with changes
"""Function type for filtering handoff inputs.
HandoffInputFilter = Callable[
[HandoffInputData],
MaybeAwaitable[HandoffInputData]
]Usage example:
from agents import Agent, handoff, HandoffInputData
async def filter_sensitive_data(data: HandoffInputData) -> HandoffInputData:
"""Remove sensitive information before handoff."""
# Filter history
filtered_history = [
item for item in data.input_history
if not contains_sensitive_info(item)
]
return data.clone(input_history=tuple(filtered_history))
specialist_agent = Agent(
name="Specialist",
instructions="Handle complex cases."
)
filtered_handoff = handoff(
specialist_agent,
input_filter=filter_sensitive_data
)
main_agent = Agent(
name="Main Agent",
handoffs=[filtered_handoff]
)Functions for managing conversation history during handoffs.
def nest_handoff_history() -> list[TResponseInputItem]:
"""
Nest handoff history into single message.
Wraps the entire conversation history in a single message
to reduce token usage and improve context management.
Returns:
- list[TResponseInputItem]: Nested history
"""
def default_handoff_history_mapper() -> list[TResponseInputItem]:
"""
Default history mapper for handoffs.
Returns:
- list[TResponseInputItem]: Mapped history
"""Type alias:
HandoffHistoryMapper = Callable[
[list[TResponseInputItem]],
list[TResponseInputItem]
]Usage example:
from agents import Agent, RunConfig, nest_handoff_history
# Global configuration
config = RunConfig(
nest_handoff_history=True,
handoff_history_mapper=nest_handoff_history
)
# Run with history management
result = Runner.run_sync(
triage_agent,
"I need help",
run_config=config
)
# Per-handoff configuration
specialist_handoff = handoff(
specialist_agent,
nest_handoff_history=True
)Global functions for managing conversation history presentation.
def get_conversation_history_wrappers() -> tuple[str, str]:
"""
Get conversation history wrappers.
Returns:
- tuple[str, str]: Opening and closing wrapper strings
"""
def set_conversation_history_wrappers(opening: str, closing: str) -> None:
"""
Set conversation history wrappers.
Parameters:
- opening: Opening wrapper string
- closing: Closing wrapper string
"""
def reset_conversation_history_wrappers() -> None:
"""Reset conversation history wrappers to defaults."""Usage example:
from agents import set_conversation_history_wrappers, reset_conversation_history_wrappers
# Customize history presentation
set_conversation_history_wrappers(
opening="<conversation_history>",
closing="</conversation_history>"
)
# Run agents with custom wrappers
result = Runner.run_sync(agent, "Hello")
# Reset to defaults
reset_conversation_history_wrappers()Enable or disable handoffs based on context.
from agents import Agent, handoff
def should_enable_expert(ctx):
"""Enable expert handoff only for premium users."""
return ctx.context.user.tier == "premium"
expert_agent = Agent(
name="Expert Agent",
instructions="Provide expert assistance."
)
expert_handoff = handoff(
expert_agent,
is_enabled=should_enable_expert
)
agent = Agent(
name="Assistant",
handoffs=[expert_handoff]
)Use Pydantic models for structured handoff parameters.
from agents import Agent, handoff
from pydantic import BaseModel
class EscalationInput(BaseModel):
reason: str
priority: str
user_id: str
escalation_agent = Agent(
name="Escalation Handler",
instructions="Handle escalated issues."
)
typed_handoff = handoff(
escalation_agent,
input_type=EscalationInput,
tool_description_override="Escalate issue with structured details"
)
agent = Agent(
name="Support Agent",
handoffs=[typed_handoff]
)Chain multiple agents with handoffs.
from agents import Agent
# Define workflow stages
intake_agent = Agent(
name="Intake",
instructions="Gather initial information."
)
analysis_agent = Agent(
name="Analysis",
instructions="Analyze the gathered information.",
handoffs=[intake_agent] # Can go back if needed
)
resolution_agent = Agent(
name="Resolution",
instructions="Provide final resolution.",
handoffs=[analysis_agent] # Can request more analysis
)
# Start with analysis stage
analysis_agent.handoffs.append(resolution_agent)
# Entry point
entry_agent = Agent(
name="Entry",
instructions="Start the workflow.",
handoffs=[intake_agent]
)
result = Runner.run_sync(entry_agent, "I need help with...")Filter conversation history based on content.
from agents import handoff, HandoffInputData
async def filter_history(data: HandoffInputData) -> HandoffInputData:
"""Keep only relevant messages."""
# Filter based on content, age, or other criteria
recent_items = data.input_history[-10:] # Last 10 messages
return data.clone(input_history=tuple(recent_items))
specialist = Agent(
name="Specialist",
instructions="Handle specialized requests."
)
filtered_handoff = handoff(
specialist,
input_filter=filter_history
)
agent = Agent(
name="General Agent",
handoffs=[filtered_handoff]
)Execute logic when handoffs occur.
from agents import handoff
import logging
logger = logging.getLogger(__name__)
async def log_handoff(ctx, agent, input_data):
"""Log handoff events for analytics."""
logger.info(f"Handoff to {agent.name}", extra={
"from_agent": ctx.agent.name,
"to_agent": agent.name,
"user_id": ctx.context.user_id
})
expert = Agent(name="Expert", instructions="Expert assistance")
monitored_handoff = handoff(
expert,
on_handoff=log_handoff
)nest_handoff_history for long conversations to manage context sizeis_enabled to dynamically control handoff availabilityon_handoff callbacks for logging and analyticsInstall with Tessl CLI
npx tessl i tessl/pypi-openai-agents