Building a Stateful Customer Service Agent with Persistent Sessions

What We Are Building

In this tutorial, we build a complete, production-ready customer service agent that:

1. Loads customer context from a database when a conversation starts
2. Accesses order history and account details through tools
3. Resolves issues across multiple conversation turns
4. Persists sessions so customers can return and continue where they left off
5. Handles handoffs between specialized agents

This is not a toy example. The patterns here reflect how real customer service AI systems are built.

Architecture Overview

The system has four layers:

flowchart TD
    START["Building a Stateful Customer Service Agent with P…"] --> A
    A["What We Are Building"]
    A --> B
    B["Architecture Overview"]
    B --> C
    C["Step 1: The Data Layer"]
    C --> D
    D["Step 2: Agent Tools"]
    D --> E
    E["Step 3: Specialized Agents"]
    E --> F
    F["Step 4: Session Layer"]
    F --> G
    G["Step 5: The Service Layer"]
    G --> H
    H["Step 6: FastAPI Integration"]
    H --> DONE["Key Takeaways"]
    style START fill:#4f46e5,stroke:#4338ca,color:#fff
    style DONE fill:#059669,stroke:#047857,color:#fff

• Session layer: SQLiteSession (swap for Redis or SQLAlchemy in production) stores conversation history
• Data layer: Customer and order data in a database (simulated here, easily swapped for real database queries)
• Agent layer: Specialized agents for different issue types with tool access
• API layer: FastAPI endpoints that tie everything together

Step 1: The Data Layer

First, define the customer and order data models. In production, these would be SQLAlchemy models or Prisma schemas. Here we use simple dataclasses:

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class OrderStatus(Enum):
    PENDING = "pending"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"
    RETURNED = "returned"

@dataclass
class OrderItem:
    product_name: str
    quantity: int
    price: float

@dataclass
class Order:
    order_id: str
    customer_id: str
    items: list[OrderItem]
    status: OrderStatus
    created_at: datetime
    tracking_number: str | None = None
    total: float = 0.0

    def __post_init__(self):
        if not self.total:
            self.total = sum(item.price * item.quantity for item in self.items)

@dataclass
class Customer:
    customer_id: str
    name: str
    email: str
    plan: str  # "free", "pro", "enterprise"
    created_at: datetime
    lifetime_value: float = 0.0

Simulated Database

from datetime import datetime, timedelta

# Simulated customer database
CUSTOMERS: dict[str, Customer] = {
    "cust_001": Customer(
        customer_id="cust_001",
        name="Sarah Chen",
        email="[email protected]",
        plan="pro",
        created_at=datetime(2025, 3, 15),
        lifetime_value=2400.00,
    ),
    "cust_002": Customer(
        customer_id="cust_002",
        name="Marcus Johnson",
        email="[email protected]",
        plan="enterprise",
        created_at=datetime(2024, 8, 1),
        lifetime_value=18500.00,
    ),
}

ORDERS: dict[str, list[Order]] = {
    "cust_001": [
        Order(
            order_id="ord_1001",
            customer_id="cust_001",
            items=[
                OrderItem("Pro Plan - Annual", 1, 199.99),
                OrderItem("Extra Seat Pack (5)", 1, 99.99),
            ],
            status=OrderStatus.DELIVERED,
            created_at=datetime.now() - timedelta(days=30),
        ),
        Order(
            order_id="ord_1002",
            customer_id="cust_001",
            items=[OrderItem("API Credits - 10K", 1, 49.99)],
            status=OrderStatus.PENDING,
            created_at=datetime.now() - timedelta(days=2),
            tracking_number=None,
        ),
    ],
    "cust_002": [
        Order(
            order_id="ord_2001",
            customer_id="cust_002",
            items=[OrderItem("Enterprise Plan - Annual", 1, 999.99)],
            status=OrderStatus.DELIVERED,
            created_at=datetime.now() - timedelta(days=90),
        ),
    ],
}

def get_customer(customer_id: str) -> Customer | None:
    return CUSTOMERS.get(customer_id)

def get_orders(customer_id: str) -> list[Order]:
    return ORDERS.get(customer_id, [])

def get_order(order_id: str) -> Order | None:
    for orders in ORDERS.values():
        for order in orders:
            if order.order_id == order_id:
                return order
    return None

Step 2: Agent Tools

Define the tools that agents use to access customer data and take actions:

from agents import function_tool

@function_tool
def lookup_customer(customer_id: str) -> str:
    """Look up a customer by their ID. Returns customer details."""
    customer = get_customer(customer_id)
    if not customer:
        return f"No customer found with ID {customer_id}"
    return (
        f"Customer: {customer.name}\n"
        f"Email: {customer.email}\n"
        f"Plan: {customer.plan}\n"
        f"Member since: {customer.created_at.strftime('%Y-%m-%d')}\n"
        f"Lifetime value: {customer.lifetime_value:.2f} USD"
    )

@function_tool
def lookup_orders(customer_id: str) -> str:
    """Look up all orders for a customer."""
    orders = get_orders(customer_id)
    if not orders:
        return f"No orders found for customer {customer_id}"

    lines = []
    for order in orders:
        items_str = ", ".join(f"{i.product_name} (x{i.quantity})" for i in order.items)
        lines.append(
            f"Order {order.order_id}: {items_str} | "
            f"{order.total:.2f} USD | Status: {order.status.value} | "
            f"Date: {order.created_at.strftime('%Y-%m-%d')}"
        )
    return "\n".join(lines)

@function_tool
def lookup_order_details(order_id: str) -> str:
    """Get detailed information about a specific order."""
    order = get_order(order_id)
    if not order:
        return f"No order found with ID {order_id}"

    lines = [
        f"Order ID: {order.order_id}",
        f"Status: {order.status.value}",
        f"Date: {order.created_at.strftime('%Y-%m-%d %H:%M')}",
        f"Tracking: {order.tracking_number or 'Not available'}",
        f"Items:",
    ]
    for item in order.items:
        lines.append(f"  - {item.product_name}: {item.quantity} x {item.price:.2f} USD")
    lines.append(f"Total: {order.total:.2f} USD")

    return "\n".join(lines)

@function_tool
def issue_refund(order_id: str, reason: str) -> str:
    """Issue a refund for an order. Requires order ID and reason."""
    order = get_order(order_id)
    if not order:
        return f"Cannot refund: order {order_id} not found"
    if order.status == OrderStatus.CANCELLED:
        return f"Order {order_id} is already cancelled/refunded"

    # In production, this would call your payment processor
    order.status = OrderStatus.RETURNED
    return (
        f"Refund of {order.total:.2f} USD issued for order {order_id}. "
        f"Reason: {reason}. "
        f"The refund will appear in 3-5 business days."
    )

@function_tool
def cancel_order(order_id: str) -> str:
    """Cancel a pending order."""
    order = get_order(order_id)
    if not order:
        return f"Cannot cancel: order {order_id} not found"
    if order.status != OrderStatus.PENDING:
        return f"Cannot cancel order {order_id}: status is {order.status.value} (only pending orders can be cancelled)"

    order.status = OrderStatus.CANCELLED
    return f"Order {order_id} has been cancelled. Any charges will be reversed."

@function_tool
def update_customer_plan(customer_id: str, new_plan: str) -> str:
    """Change a customer's subscription plan."""
    customer = get_customer(customer_id)
    if not customer:
        return f"Customer {customer_id} not found"

    valid_plans = ["free", "pro", "enterprise"]
    if new_plan not in valid_plans:
        return f"Invalid plan: {new_plan}. Valid plans: {', '.join(valid_plans)}"

    old_plan = customer.plan
    customer.plan = new_plan
    return f"Plan changed from {old_plan} to {new_plan} for {customer.name}"

Step 3: Specialized Agents

Create agents that specialize in different issue types:

flowchart TD
    ROOT["Building a Stateful Customer Service Agent w…"] 
    ROOT --> P0["Step 1: The Data Layer"]
    P0 --> P0C0["Simulated Database"]
    ROOT --> P1["Step 7: Running the Full System"]
    P1 --> P1C0["Interactive CLI Demo"]
    P1 --> P1C1["Sample Conversation"]
    style ROOT fill:#4f46e5,stroke:#4338ca,color:#fff
    style P0 fill:#e0e7ff,stroke:#6366f1,color:#1e293b
    style P1 fill:#e0e7ff,stroke:#6366f1,color:#1e293b

from agents import Agent, handoff

# Billing specialist
billing_agent = Agent(
    name="BillingSpecialist",
    instructions="""You are a billing specialist. You handle:
    - Refund requests
    - Charge disputes
    - Plan changes and upgrades/downgrades
    - Payment method issues

    Always look up the customer and their orders before taking action.
    Confirm the specific order and amount before issuing refunds.
    Be empathetic but efficient. Follow company policy:
    - Refunds under $100: approve immediately
    - Refunds $100-$500: approve with documented reason
    - Refunds over $500: inform the customer it needs manager approval

    You have access to the full conversation history. Never ask the
    customer to repeat information already discussed.""",
    tools=[lookup_customer, lookup_orders, lookup_order_details,
           issue_refund, cancel_order, update_customer_plan],
)

# Technical support specialist
technical_agent = Agent(
    name="TechnicalSupport",
    instructions="""You are a technical support specialist. You handle:
    - API issues and error codes
    - Integration troubleshooting
    - Feature questions and how-to guidance
    - Bug reports

    Look up the customer's plan to understand what features they
    have access to. Check their order history for relevant purchases.

    For bug reports, collect: steps to reproduce, expected behavior,
    actual behavior, and any error messages.

    You have access to the full conversation history.""",
    tools=[lookup_customer, lookup_orders],
)

# Triage agent — routes to specialists
triage_agent = Agent(
    name="CustomerServiceTriage",
    instructions="""You are the first point of contact for customer service.
    Your job is to:
    1. Greet the customer warmly
    2. Identify the customer (ask for customer ID or email)
    3. Understand their issue
    4. Route to the appropriate specialist

    Use transfer_to_billing for: refunds, charges, plan changes, payments.
    Use transfer_to_technical for: bugs, API issues, feature questions, integration help.

    Always look up the customer first to personalize the interaction.
    If the customer's plan is "enterprise", mention that they have priority support.""",
    tools=[lookup_customer, lookup_orders],
    handoffs=[
        handoff(billing_agent,
                tool_name="transfer_to_billing",
                tool_description="Transfer to billing specialist for payment and subscription issues"),
        handoff(technical_agent,
                tool_name="transfer_to_technical",
                tool_description="Transfer to technical support for product and API issues"),
    ],
)

Step 4: Session Layer

Set up persistent sessions so customers can continue conversations:

from agents.extensions.sessions import SQLiteSession, SessionSettings

# In production, use RedisSession or SQLAlchemySession
session = SQLiteSession(db_path="./customer_service.db")
settings = SessionSettings(limit=60)

Step 5: The Service Layer

Create a service class that ties everything together:

from agents import Runner
from dataclasses import dataclass, field

@dataclass
class ConversationState:
    session_id: str
    customer_id: str | None = None
    current_agent_name: str = "CustomerServiceTriage"
    turns: int = 0

class CustomerServiceSystem:
    """Complete customer service system with stateful conversations."""

    def __init__(self):
        self.session = SQLiteSession(db_path="./customer_service.db")
        self.settings = SessionSettings(limit=60)
        self.conversations: dict[str, ConversationState] = {}
        self.agents = {
            "CustomerServiceTriage": triage_agent,
            "BillingSpecialist": billing_agent,
            "TechnicalSupport": technical_agent,
        }

    def get_or_create_conversation(self, session_id: str) -> ConversationState:
        """Get existing conversation state or create new one."""
        if session_id not in self.conversations:
            self.conversations[session_id] = ConversationState(session_id=session_id)
        return self.conversations[session_id]

    async def handle_message(self, session_id: str, message: str) -> dict:
        """Process a customer message and return the response."""
        conv = self.get_or_create_conversation(session_id)
        current_agent = self.agents[conv.current_agent_name]

        result = await Runner.run(
            current_agent,
            message,
            session=self.session,
            session_id=session_id,
            session_settings=self.settings,
        )

        # Track which agent handled the response
        new_agent_name = result.last_agent.name
        if new_agent_name != conv.current_agent_name:
            print(f"Handoff: {conv.current_agent_name} -> {new_agent_name}")
            conv.current_agent_name = new_agent_name

        conv.turns += 1

        return {
            "response": result.final_output,
            "agent": new_agent_name,
            "session_id": session_id,
            "turn": conv.turns,
        }

    async def get_conversation_summary(self, session_id: str) -> dict:
        """Get a summary of the conversation state."""
        conv = self.conversations.get(session_id)
        if not conv:
            return {"error": "Conversation not found"}

        items = await self.session.get_items(session_id)
        return {
            "session_id": session_id,
            "customer_id": conv.customer_id,
            "current_agent": conv.current_agent_name,
            "total_turns": conv.turns,
            "history_items": len(items),
        }

Step 6: FastAPI Integration

Expose the service through a REST API:

flowchart LR
    S0["Step 1: The Data Layer"]
    S0 --> S1
    S1["Step 2: Agent Tools"]
    S1 --> S2
    S2["Step 3: Specialized Agents"]
    S2 --> S3
    S3["Step 4: Session Layer"]
    S3 --> S4
    S4["Step 5: The Service Layer"]
    S4 --> S5
    S5["Step 6: FastAPI Integration"]
    style S0 fill:#4f46e5,stroke:#4338ca,color:#fff
    style S5 fill:#059669,stroke:#047857,color:#fff

from fastapi import FastAPI
from pydantic import BaseModel
import uuid

app = FastAPI(title="Customer Service API")
service = CustomerServiceSystem()

class ChatRequest(BaseModel):
    session_id: str | None = None
    message: str

class ChatResponse(BaseModel):
    session_id: str
    response: str
    agent: str
    turn: int

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """Send a message to the customer service system."""
    session_id = request.session_id or f"cs-{uuid.uuid4().hex[:12]}"

    result = await service.handle_message(session_id, request.message)

    return ChatResponse(
        session_id=result["session_id"],
        response=result["response"],
        agent=result["agent"],
        turn=result["turn"],
    )

@app.get("/conversation/{session_id}")
async def get_conversation(session_id: str):
    """Get conversation summary and metadata."""
    return await service.get_conversation_summary(session_id)

@app.post("/conversation/{session_id}/reset")
async def reset_conversation(session_id: str):
    """Reset a conversation — clear session and state."""
    await service.session.clear_session(session_id)
    if session_id in service.conversations:
        del service.conversations[session_id]
    return {"status": "reset", "session_id": session_id}

Step 7: Running the Full System

Interactive CLI Demo

import asyncio

async def interactive_demo():
    service = CustomerServiceSystem()
    session_id = f"demo-{uuid.uuid4().hex[:8]}"
    print(f"Customer Service Demo | Session: {session_id}")
    print("Type 'quit' to exit, 'summary' for conversation stats\n")

    while True:
        user_input = input("Customer: ").strip()
        if not user_input:
            continue
        if user_input.lower() == "quit":
            break
        if user_input.lower() == "summary":
            summary = await service.get_conversation_summary(session_id)
            print(f"\nConversation Summary: {summary}\n")
            continue

        result = await service.handle_message(session_id, user_input)
        print(f"\n[{result['agent']}]: {result['response']}\n")

asyncio.run(interactive_demo())

Sample Conversation

Here is what a typical interaction looks like:

Customer: Hi, I'm customer cust_001 and I need help with a billing issue.

[CustomerServiceTriage]: Hello Sarah! I can see you're on our Pro plan and
have been a valued member since March 2025. I'd be happy to help with your
billing issue. Let me transfer you to our billing specialist.

[BillingSpecialist]: Hi Sarah! I can see your account and recent orders.
You have two orders on file:
- Order ord_1001: Pro Plan Annual + Extra Seats ($299.98) — Delivered
- Order ord_1002: API Credits 10K ($49.99) — Pending

What billing issue can I help you with?

Customer: I'd like a refund for the API credits order, I haven't used them yet.

[BillingSpecialist]: I can help with that. Let me process a refund for order
ord_1002 (API Credits - 10K, $49.99). Since this is under $100, I can approve
it immediately.

The refund of $49.99 has been issued. It will appear on your statement in
3-5 business days. Is there anything else I can help with?

Customer: Can you also downgrade my plan to free?

[BillingSpecialist]: I've updated your plan from Pro to Free. Please note
that some features from the Pro plan will no longer be available. Is there
anything else?

The key points of this interaction: the triage agent looked up the customer and personalized the greeting, the billing agent saw the full history from triage, the refund followed the under-$100 auto-approve policy, and the plan change was handled in the same session with full context.

Production Considerations

When moving this to production, swap the simulated database for real queries, use RedisSession or SQLAlchemySession instead of SQLite, add authentication middleware to map API requests to customer IDs, implement rate limiting per session, add observability with structured logging for every tool call and handoff, and set up session TTL to automatically clean up abandoned conversations.

The patterns in this tutorial — tool-equipped agents, handoffs, shared sessions, and layered architecture — form the foundation of real customer service AI systems.

Sources:

• https://openai.github.io/openai-agents-python/sessions/
• https://openai.github.io/openai-agents-python/handoffs/
• https://openai.github.io/openai-agents-python/tools/