In the previous tutorial, we covered the basics of the OpenAI Agents SDK: creating agents, using tools, and setting up simple handoffs. Now let’s build something more practical: a customer service system for an airline.
This project demonstrates several real-world patterns:
- Shared context across agents using Pydantic models
- Bidirectional handoffs where agents can transfer back and forth
- Specialized agents with domain-specific tools and routines
By the end, you’ll have a working multi-agent system that handles FAQ queries and seat changes.
Architecture Overview#
Our system has three agents:
- Triage Agent - Entry point that routes requests to specialists
- FAQ Agent - Answers questions about baggage, seats, wifi, etc.
- Seat Booking Agent - Handles seat changes
The key difference from the basic handoff example: each agent can hand off back to the triage agent if they can’t handle a request. This creates a flexible routing system.
┌─────────────────┐
│ Triage Agent │
│ (Entry Point) │
└────────┬────────┘
│
┌────┴────┐
▼ ▼
┌───────┐ ┌──────────┐
│ FAQ │ │ Seat │
│ Agent │ │ Booking │
└───┬───┘ └────┬─────┘
│ │
└────┬─────┘
▼
Back to Triage
Setup#
Install the SDK with LiteLLM support:
pip install "openai-agents[litellm]" python-dotenv
Set up your environment:
import os
import getpass
from dotenv import load_dotenv
load_dotenv()
if "HF_TOKEN" not in os.environ:
os.environ["HF_TOKEN"] = getpass.getpass("Enter your HuggingFace token: ")
Shared Context#
All agents share context through a Pydantic model. This is how information flows between agents during handoffs:
from pydantic import BaseModel
class AirlineAgentContext(BaseModel):
passenger_name: str | None = None
confirmation_number: str | None = None
seat_number: str | None = None
flight_number: str | None = None
When you run the agent system, you initialize this context with known information (like the customer’s flight number). As agents interact with the customer, they update the context with new information (like a confirmation number).
Creating the Tools#
We need two tools: one for FAQ lookups, another for seat updates.
FAQ Lookup Tool#
This simulates a knowledge base search. In production, you’d connect this to a vector database or search API:
from agents import function_tool
@function_tool(
name_override="faq_lookup_tool",
description_override="Lookup frequently asked questions."
)
async def faq_lookup_tool(question: str) -> str:
question_lower = question.lower()
if any(keyword in question_lower for keyword in ["bag", "baggage", "luggage"]):
return (
"You are allowed to bring one bag on the plane. "
"It must be under 50 pounds and 22 x 14 x 9 inches."
)
elif any(keyword in question_lower for keyword in ["seat", "seats", "seating"]):
return (
"There are 120 seats on the plane. "
"22 business class, 98 economy. "
"Exit rows: 4 and 16. Rows 5-8 are Economy Plus with extra legroom."
)
elif any(keyword in question_lower for keyword in ["wifi", "internet"]):
return "We have free wifi on the plane, join Airline-Wifi"
return "I'm sorry, I don't know the answer to that question."
Seat Update Tool#
This tool accesses the shared context to update seat information. Notice how it uses RunContextWrapper to read and modify the context:
from agents import function_tool, RunContextWrapper
@function_tool
async def update_seat(
ctx: RunContextWrapper[AirlineAgentContext],
confirmation_number: str,
new_seat: str
) -> str:
"""
Update the seat for a given confirmation number.
Args:
confirmation_number: The confirmation number for the flight.
new_seat: The new seat to update to.
"""
# Update context with customer input
ctx.context.confirmation_number = confirmation_number
ctx.context.seat_number = new_seat
# Verify flight number exists (set during initialization)
assert ctx.context.flight_number is not None, "Flight number is required"
return f"Updated seat to {new_seat} for confirmation number {confirmation_number}"
The RunContextWrapper[AirlineAgentContext] type annotation tells the SDK this tool needs access to the shared context. The SDK handles passing the context automatically.
Building the Agents#
First, set up the model. We’re using a Hugging Face model through LiteLLM:
from agents.extensions.models.litellm_model import LitellmModel
model = LitellmModel(
model="huggingface/nscale/Qwen/Qwen3-8B",
api_key=os.environ["HF_TOKEN"],
)
FAQ Agent#
The FAQ agent has a clear routine: identify the question, look it up, and transfer back if it can’t help:
from agents import Agent
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX
faq_agent = Agent[AirlineAgentContext](
name="FAQ Agent",
handoff_description="A helpful agent that can answer questions about the airline.",
instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are an FAQ agent. If you are speaking to a customer, you probably were transferred to from the triage agent.
Use the following routine to support the customer.
# Routine
1. Identify the last question asked by the customer.
2. Use the faq lookup tool to answer the question. Do not rely on your own knowledge.
3. If you cannot answer the question, transfer back to the triage agent.""",
tools=[faq_lookup_tool],
model=model,
)
Note the RECOMMENDED_PROMPT_PREFIX - this is a built-in prompt that helps agents handle handoffs correctly.
Seat Booking Agent#
Similar structure, but with a transactional routine:
seat_booking_agent = Agent[AirlineAgentContext](
name="Seat Booking Agent",
handoff_description="A helpful agent that can update a seat on a flight.",
instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a seat booking agent. If you are speaking to a customer, you probably were transferred to from the triage agent.
Use the following routine to support the customer.
# Routine
1. Ask for their confirmation number.
2. Ask the customer what their desired seat number is.
3. Use the update seat tool to update the seat on the flight.
If the customer asks a question that is not related to the routine, transfer back to the triage agent.""",
tools=[update_seat],
model=model,
)
Triage Agent#
The triage agent has no tools - it only routes to specialists:
triage_agent = Agent[AirlineAgentContext](
name="Triage Agent",
handoff_description="A triage agent that can delegate a customer's request to the appropriate agent.",
instructions=(
f"{RECOMMENDED_PROMPT_PREFIX} "
"You are a helpful triaging agent. You can use your tools to delegate questions to other appropriate agents."
),
handoffs=[faq_agent, seat_booking_agent],
model=model,
)
Connecting Bidirectional Handoffs#
Here’s the crucial step - allowing specialists to hand back to triage:
faq_agent.handoffs.append(triage_agent)
seat_booking_agent.handoffs.append(triage_agent)
This creates a flexible system where any agent can route to any other through triage.
Running the System#
Create a session for conversation persistence and run:
import uuid
from agents import Runner, SQLiteSession
# Create unique session
session_id = str(uuid.uuid4())
session = SQLiteSession(session_id)
# Initialize context with known information
context = AirlineAgentContext(
flight_number="FLT-123",
seat_number="A12",
passenger_name="John Doe",
)
# Run the agent
result = await Runner.run(
triage_agent,
input="can i change my seat?",
session=session,
context=context,
)
print(result.final_output)
Output:
I can help you change your seat. Could you please provide your confirmation number?
The triage agent routes to the seat booking agent, which then asks for the confirmation number.
Visualizing the Graph#
The SDK includes a visualization tool:
pip install "openai-agents[viz]"
from agents.extensions.visualization import draw_graph
draw_graph(triage_agent)
This generates a graph showing all agents and their handoff connections.
Interactive Demo Loop#
For testing, use the built-in demo loop:
from agents import run_demo_loop
async def main():
context = AirlineAgentContext(
flight_number="FLT-123",
seat_number="A12",
passenger_name="John Doe",
)
await run_demo_loop(triage_agent, context=context)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
This gives you an interactive terminal to test conversations.
Key Patterns#
1. Typed Context#
Using Agent[AirlineAgentContext] ensures type safety. The context is available to all tools via RunContextWrapper[AirlineAgentContext].
2. Routine-Based Instructions#
Each agent has a numbered routine. This helps the LLM follow a consistent process:
# Routine
1. Ask for confirmation number
2. Ask for desired seat
3. Use update_seat tool
3. Handoff Descriptions#
The handoff_description parameter tells other agents what this agent specializes in. The triage agent uses these descriptions to route appropriately.
4. Bidirectional Flow#
By adding handoffs in both directions, agents can gracefully handle edge cases by transferring back to triage.
Recap#
You built a multi-agent customer service system with:
- Shared context that persists across agent handoffs
- Specialized tools that access and modify context
- Bidirectional handoffs for flexible routing
- Session persistence for multi-turn conversations
This pattern scales well. Add new specialist agents by creating them with their tools and routines, then connecting them to the triage agent.
Full Code#
from agents import function_tool, RunContextWrapper, Agent, Runner, SQLiteSession
from agents.extensions.models.litellm_model import LitellmModel
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX
from pydantic import BaseModel
import os
import uuid
# Context shared across all agents
class AirlineAgentContext(BaseModel):
passenger_name: str | None = None
confirmation_number: str | None = None
seat_number: str | None = None
flight_number: str | None = None
# Tools
@function_tool(
name_override="faq_lookup_tool",
description_override="Lookup frequently asked questions."
)
async def faq_lookup_tool(question: str) -> str:
question_lower = question.lower()
if any(kw in question_lower for kw in ["bag", "baggage", "luggage"]):
return "One bag allowed, under 50 lbs and 22x14x9 inches."
elif any(kw in question_lower for kw in ["seat", "seats"]):
return "120 seats: 22 business, 98 economy. Exit rows: 4, 16. Economy Plus: rows 5-8."
elif any(kw in question_lower for kw in ["wifi", "internet"]):
return "Free wifi available, join Airline-Wifi"
return "I don't know the answer to that question."
@function_tool
async def update_seat(
ctx: RunContextWrapper[AirlineAgentContext],
confirmation_number: str,
new_seat: str
) -> str:
"""Update the seat for a given confirmation number."""
ctx.context.confirmation_number = confirmation_number
ctx.context.seat_number = new_seat
assert ctx.context.flight_number is not None
return f"Updated seat to {new_seat} for confirmation {confirmation_number}"
# Model
model = LitellmModel(
model="huggingface/nscale/Qwen/Qwen3-8B",
api_key=os.environ["HF_TOKEN"],
)
# Agents
faq_agent = Agent[AirlineAgentContext](
name="FAQ Agent",
handoff_description="Answers questions about the airline.",
instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are an FAQ agent.
# Routine
1. Identify the question.
2. Use faq_lookup_tool to answer. Don't use your own knowledge.
3. If you can't answer, transfer to triage agent.""",
tools=[faq_lookup_tool],
model=model,
)
seat_booking_agent = Agent[AirlineAgentContext](
name="Seat Booking Agent",
handoff_description="Updates seats on flights.",
instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
You are a seat booking agent.
# Routine
1. Ask for confirmation number.
2. Ask for desired seat number.
3. Use update_seat tool.
If unrelated question, transfer to triage agent.""",
tools=[update_seat],
model=model,
)
triage_agent = Agent[AirlineAgentContext](
name="Triage Agent",
handoff_description="Routes requests to appropriate agents.",
instructions=f"{RECOMMENDED_PROMPT_PREFIX} Route questions to appropriate agents.",
handoffs=[faq_agent, seat_booking_agent],
model=model,
)
# Bidirectional handoffs
faq_agent.handoffs.append(triage_agent)
seat_booking_agent.handoffs.append(triage_agent)
# Run
async def main():
session = SQLiteSession(str(uuid.uuid4()))
context = AirlineAgentContext(
flight_number="FLT-123",
seat_number="A12",
passenger_name="John Doe",
)
result = await Runner.run(
triage_agent,
input="Can I change my seat?",
session=session,
context=context,
)
print(result.final_output)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
