Introduction to AI-Powered Customer Service
An AI-powered customer service department is a system where multiple specialized AI agents collaborate to handle incoming customer inquiries — classifying intent, retrieving knowledge, resolving common issues, and escalating complex cases to human agents when necessary. Unlike a simple chatbot that follows a fixed script, an agentic system reasons about the best course of action, selects the right tools, and orchestrates multiple steps to achieve a resolution.
Think of it as a digital team where each agent has a distinct role: one triages incoming tickets, another searches the knowledge base, a third handles billing queries, and a supervisor agent coordinates handoffs. These agents use large language models (LLMs) as their reasoning engine, augmented with tools like databases, APIs, and internal documentation.
Why AI Agents for Customer Service Matter
Traditional customer service departments face three persistent challenges: volume (more tickets than agents can handle), consistency (different agents give different answers), and latency (customers wait hours or days for resolutions). AI agents address all three simultaneously.
Key business drivers include:
- 24/7 availability — agents never sleep, handling off-hours inquiries instantly
- Cost reduction — organizations report 40–60% deflection of tier-1 tickets to AI
- Faster resolution — AI agents resolve common issues in seconds rather than hours
- Scalability — the system handles 10 or 10,000 concurrent conversations without degradation
- Auditability — every AI decision is logged, making compliance and improvement easier than with human-only teams
- Human focus — human agents concentrate on high-value, complex cases where empathy and creativity matter
Core Architecture Overview
A production-grade AI customer service department typically has these components:
- Intent Classifier Agent — categorizes the customer's request and determines urgency
- Knowledge Retrieval Agent — searches internal docs, FAQs, and policy databases
- Specialist Agents — domain-specific agents for billing, technical support, returns, etc.
- Ticket Router / Orchestrator — coordinates between agents and decides when to escalate
- Human Escalation Queue — structured handoff with full context preservation
- Memory & Session Store — maintains conversation state across multiple turns
The orchestration pattern follows a tool-calling loop: the orchestrator receives the customer message, calls the classifier, decides which specialist to invoke, aggregates results, and returns a coherent response. If confidence drops below a threshold or the customer explicitly requests a human, the system escalates with a complete transcript.
Building Your First AI Customer Service Agent
Setting Up the Environment
You'll need Python 3.10+, an LLM provider (we'll use OpenAI's API as an example, but the pattern works with any provider), and a vector database for knowledge retrieval. Start by installing dependencies:
# requirements.txt
openai>=1.0.0
chromadb>=0.4.0
pydantic>=2.0.0
fastapi>=0.100.0
uvicorn>=0.23.0
python-dotenv>=1.0.0
Set up your project structure:
customer_service_ai/
├── agents/
│ ├── __init__.py
│ ├── classifier.py # Intent classification agent
│ ├── knowledge.py # Knowledge retrieval agent
│ ├── billing.py # Billing specialist agent
│ ├── technical.py # Technical support agent
│ └── orchestrator.py # Main coordinator
├── tools/
│ ├── __init__.py
│ ├── ticket_api.py # Ticket system integration
│ ├── kb_search.py # Knowledge base search tool
│ └── human_handoff.py # Escalation tool
├── memory/
│ ├── __init__.py
│ └── session_store.py # Conversation memory
├── config.py
├── main.py
└── .env
Creating the Intent Classifier
The classifier agent analyzes the customer's message and returns a structured classification. This is the entry point for every conversation — it determines which specialist should handle the request and at what priority level.
# agents/classifier.py
import json
from openai import OpenAI
from pydantic import BaseModel
from typing import Literal, Optional
class IntentClassification(BaseModel):
category: Literal["billing", "technical", "returns", "general_inquiry", "complaint"]
urgency: Literal["low", "medium", "high", "critical"]
sentiment: Literal["neutral", "frustrated", "angry", "satisfied"]
summary: str
requires_human: bool
reason_for_human: Optional[str] = None
class IntentClassifier:
def __init__(self, client: OpenAI):
self.client = client
self.system_prompt = """You are an expert customer service triage agent.
Your job is to analyze incoming customer messages and classify them accurately.
Consider these guidelines:
- billing: Payment issues, refunds, charges, invoices, subscription changes
- technical: Product malfunctions, errors, setup help, integration problems
- returns: Return requests, exchange inquiries, warranty claims, damaged items
- general_inquiry: Product questions, pricing, availability, policy questions
- complaint: General dissatisfaction not tied to a specific category above
Urgency levels:
- critical: Service outage, data loss, security breach, payment blocked
- high: Product unusable, deadline-sensitive, repeated contact
- medium: Standard issue, workaround exists
- low: General question, nice-to-have request
Flag requires_human when:
- The customer explicitly demands a human agent
- Legal or compliance issues are mentioned
- The situation involves potential fraud or account takeover
- Sentiment is angry combined with critical urgency"""
def classify(self, message: str, conversation_history: str = "") -> IntentClassification:
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"Previous conversation:\n{conversation_history}\n\nNew message:\n{message}"}
],
tools=[{
"type": "function",
"function": {
"name": "classify_intent",
"description": "Classify the customer's intent",
"parameters": {
"type": "object",
"properties": {
"category": {"type": "string", "enum": ["billing", "technical", "returns", "general_inquiry", "complaint"]},
"urgency": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
"sentiment": {"type": "string", "enum": ["neutral", "frustrated", "angry", "satisfied"]},
"summary": {"type": "string", "description": "One-sentence summary of the issue"},
"requires_human": {"type": "boolean"},
"reason_for_human": {"type": "string", "description": "Required if requires_human is true"}
},
"required": ["category", "urgency", "sentiment", "summary", "requires_human"]
}
}
}],
tool_choice={"type": "function", "function": {"name": "classify_intent"}},
temperature=0.1
)
tool_call = response.choices[0].message.tool_calls[0]
args = json.loads(tool_call.function.arguments)
return IntentClassification(**args)
Building the Knowledge Base Agent
The knowledge agent retrieves relevant information from your internal documentation. We'll use ChromaDB for vector storage and implement a retrieval-augmented generation (RAG) pattern that cites sources.
# agents/knowledge.py
import chromadb
from openai import OpenAI
from typing import List, Dict
class KnowledgeRetrievalAgent:
def __init__(self, client: OpenAI, collection_name: str = "support_docs"):
self.client = client
self.chroma_client = chromadb.PersistentClient(path="./kb_storage")
# Create or get existing collection
try:
self.collection = self.chroma_client.get_collection(collection_name)
except Exception:
self.collection = self.chroma_client.create_collection(
name=collection_name,
metadata={"hnsw:space": "cosine"}
)
def index_documents(self, documents: List[Dict[str, str]]):
"""Index documents into the vector store.
Each document should have: id, content, metadata (title, category, url)"""
ids = [doc["id"] for doc in documents]
contents = [doc["content"] for doc in documents]
metadatas = [doc.get("metadata", {}) for doc in documents]
# Generate embeddings in batches
embeddings = []
for i in range(0, len(contents), 20):
batch = contents[i:i+20]
response = self.client.embeddings.create(
model="text-embedding-3-small",
input=batch
)
embeddings.extend([e.embedding for e in response.data])
self.collection.add(
ids=ids,
embeddings=embeddings,
documents=contents,
metadatas=metadatas
)
def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""Search the knowledge base for relevant documents."""
query_embedding = self.client.embeddings.create(
model="text-embedding-3-small",
input=[query]
).data[0].embedding
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k
)
documents = []
for i, doc_id in enumerate(results["ids"][0]):
documents.append({
"id": doc_id,
"content": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"relevance_score": 1 - results["distances"][0][i]
if results["distances"] else None
})
return documents
def generate_answer(self, query: str, retrieved_docs: List[Dict]) -> Dict:
"""Generate a contextual answer based on retrieved documents."""
context_parts = []
for i, doc in enumerate(retrieved_docs):
source = doc["metadata"].get("title", f"Document {i+1}")
context_parts.append(f"[Source {i+1}: {source}]\n{doc['content']}")
context = "\n\n".join(context_parts)
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": f"""You are a helpful customer support knowledge agent.
Answer the customer's question using ONLY the provided context documents.
If the context doesn't contain the answer, say: "I don't have enough information to answer this accurately. Let me connect you with a specialist."
Always cite sources using [Source N] notation.
Be concise but complete."""},
{"role": "user", "content": f"Context documents:\n{context}\n\nCustomer question: {query}"}
],
temperature=0.2
)
answer = response.choices[0].message.content
# Extract cited sources
cited_sources = []
for doc in retrieved_docs:
if f"[Source {retrieved_docs.index(doc)+1}]" in answer:
cited_sources.append(doc["metadata"])
return {
"answer": answer,
"sources": cited_sources,
"has_answer": "don't have enough information" not in answer.lower()
}
Implementing the Specialist Agents
Specialist agents handle domain-specific tasks. Each has access to relevant tools — the billing agent can look up invoices and process refunds, while the technical agent can query system logs and run diagnostics. Here's the billing specialist:
# agents/billing.py
import json
from openai import OpenAI
from tools.ticket_api import TicketAPI
class BillingAgent:
def __init__(self, client: OpenAI, ticket_api: TicketAPI):
self.client = client
self.ticket_api = ticket_api
self.system_prompt = """You are a billing specialist agent for a SaaS company.
You have access to these tools:
- lookup_invoice: Find invoice details by customer email or invoice ID
- list_charges: List recent charges for a customer account
- process_refund: Issue a refund for a specific charge (requires supervisor approval for amounts > $100)
- check_subscription: View current subscription status and plan details
Always be empathetic about financial issues. Follow these rules:
1. Verify customer identity before discussing charges (ask for email + last 4 digits of payment method)
2. Explain charges clearly before offering refunds
3. For amounts over $100, inform the customer that refunds require review
4. Never promise refunds you cannot verify
5. When in doubt, escalate to human agent with a summary"""
def handle(self, customer_message: str, customer_context: dict, conversation_history: str) -> dict:
"""Process a billing-related customer inquiry."""
# Build context for the LLM
context = f"""Customer context:
Email: {customer_context.get('email', 'Not verified')}
Account ID: {customer_context.get('account_id', 'Not verified')}
Subscription: {customer_context.get('subscription_plan', 'Unknown')}
Conversation history:
{conversation_history}"""
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"{context}\n\nCustomer message: {customer_message}"}
]
# Define available tools for this specialist
tools = [
{
"type": "function",
"function": {
"name": "lookup_invoice",
"description": "Look up an invoice by customer email or invoice ID",
"parameters": {
"type": "object",
"properties": {
"customer_email": {"type": "string"},
"invoice_id": {"type": "string", "description": "Optional, if known"}
},
"required": ["customer_email"]
}
}
},
{
"type": "function",
"function": {
"name": "list_charges",
"description": "List recent charges for a customer account",
"parameters": {
"type": "object",
"properties": {
"account_id": {"type": "string"},
"days": {"type": "integer", "description": "Number of days to look back, default 90"}
},
"required": ["account_id"]
}
}
},
{
"type": "function",
"function": {
"name": "process_refund",
"description": "Issue a refund for a charge",
"parameters": {
"type": "object",
"properties": {
"charge_id": {"type": "string"},
"amount": {"type": "number"},
"reason": {"type": "string"}
},
"required": ["charge_id", "amount", "reason"]
}
}
},
{
"type": "function",
"function": {
"name": "escalate_to_human",
"description": "Escalate this conversation to a human agent with full context",
"parameters": {
"type": "object",
"properties": {
"reason": {"type": "string"},
"summary": {"type": "string", "description": "Detailed summary for the human agent"}
},
"required": ["reason", "summary"]
}
}
}
]
# Tool-calling loop
max_turns = 5
turn = 0
while turn < max_turns:
turn += 1
response = self.client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
temperature=0.3
)
message = response.choices[0].message
if message.tool_calls:
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
# Execute the tool
if func_name == "lookup_invoice":
result = self.ticket_api.lookup_invoice(
customer_email=func_args.get("customer_email"),
invoice_id=func_args.get("invoice_id")
)
elif func_name == "list_charges":
result = self.ticket_api.list_charges(
account_id=func_args.get("account_id"),
days=func_args.get("days", 90)
)
elif func_name == "process_refund":
amount = func_args.get("amount", 0)
if amount > 100:
result = {"status": "pending_review", "message": f"Refund of ${amount} requires supervisor approval. Case has been flagged."}
else:
result = self.ticket_api.process_refund(
charge_id=func_args.get("charge_id"),
amount=amount,
reason=func_args.get("reason", "")
)
elif func_name == "escalate_to_human":
return {
"action": "escalate",
"reason": func_args.get("reason"),
"summary": func_args.get("summary"),
"agent_type": "human"
}
# Add the tool result to the conversation
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
else:
# No tool calls — the agent is responding directly
return {
"action": "respond",
"response": message.content,
"agent_type": "billing_specialist"
}
# Exceeded max turns — escalate
return {
"action": "escalate",
"reason": "Agent exceeded maximum reasoning turns",
"summary": f"Billing agent could not resolve: {customer_message}",
"agent_type": "human"
}
Building the Technical Support Agent
The technical support agent follows the same pattern but with different tools — it can search error logs, run diagnostics, and consult the knowledge base:
# agents/technical.py
import json
from openai import OpenAI
from agents.knowledge import KnowledgeRetrievalAgent
from typing import Dict
class TechnicalSupportAgent:
def __init__(self, client: OpenAI, knowledge_agent: KnowledgeRetrievalAgent):
self.client = client
self.knowledge_agent = knowledge_agent
self.system_prompt = """You are a technical support specialist.
You help customers troubleshoot product issues, configuration problems,
and integration errors. You have access to:
- search_knowledge_base: Search internal documentation for solutions
- run_diagnostic: Run a diagnostic check on the customer's account/setup
- create_bug_report: File a bug report for the engineering team
Approach every issue methodically:
1. Understand the exact symptoms
2. Check if this is a known issue (search knowledge base)
3. Guide the customer through troubleshooting steps
4. If unresolved, gather diagnostic data and escalate if needed
Always ask clarifying questions before jumping to solutions."""
def handle(self, customer_message: str, customer_context: dict, conversation_history: str) -> Dict:
context = f"""Customer context:
Product: {customer_context.get('product', 'Unknown')}
Version: {customer_context.get('version', 'Unknown')}
Environment: {customer_context.get('environment', 'Unknown')}
Previous conversation:
{conversation_history}"""
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"{context}\n\nCustomer message: {customer_message}"}
]
tools = [
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "Search internal documentation for solutions",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"},
"top_k": {"type": "integer", "description": "Number of results, default 5"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "run_diagnostic",
"description": "Run diagnostic checks on the customer's setup",
"parameters": {
"type": "object",
"properties": {
"account_id": {"type": "string"},
"check_type": {"type": "string", "enum": ["connectivity", "configuration", "performance", "all"]}
},
"required": ["account_id", "check_type"]
}
}
},
{
"type": "function",
"function": {
"name": "escalate_to_human",
"description": "Escalate to a human technical support agent",
"parameters": {
"type": "object",
"properties": {
"reason": {"type": "string"},
"summary": {"type": "string"},
"diagnostic_data": {"type": "object"}
},
"required": ["reason", "summary"]
}
}
}
]
max_turns = 5
turn = 0
while turn < max_turns:
turn += 1
response = self.client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
temperature=0.3
)
message = response.choices[0].message
if message.tool_calls:
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
if func_name == "search_knowledge_base":
docs = self.knowledge_agent.search(
query=func_args.get("query"),
top_k=func_args.get("top_k", 5)
)
kb_result = self.knowledge_agent.generate_answer(
query=func_args.get("query"),
retrieved_docs=docs
)
result = kb_result
elif func_name == "run_diagnostic":
# Simulated diagnostic (in production, this would call real APIs)
result = {
"status": "completed",
"checks": {
"connectivity": "healthy",
"api_latency_ms": 145,
"error_rate_24h": "0.02%",
"recent_errors": ["timeout_on_webhook_endpoint_3h_ago"]
},
"recommendation": "Check webhook endpoint configuration"
}
elif func_name == "escalate_to_human":
return {
"action": "escalate",
"reason": func_args.get("reason"),
"summary": func_args.get("summary"),
"diagnostic_data": func_args.get("diagnostic_data", {}),
"agent_type": "human"
}
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
else:
return {
"action": "respond",
"response": message.content,
"agent_type": "technical_specialist"
}
return {
"action": "escalate",
"reason": "Agent exceeded maximum reasoning turns",
"summary": f"Technical agent could not resolve: {customer_message}",
"agent_type": "human"
}
Assembling the Orchestrator
The orchestrator is the central nervous system. It receives every customer message, calls the classifier, routes to the appropriate specialist, manages the conversation loop, and decides when to escalate to humans. This is the component you expose via your API.
# agents/orchestrator.py
from openai import OpenAI
from agents.classifier import IntentClassifier, IntentClassification
from agents.billing import BillingAgent
from agents.technical import TechnicalSupportAgent
from agents.knowledge import KnowledgeRetrievalAgent
from memory.session_store import SessionStore
from tools.ticket_api import TicketAPI
from typing import Dict, Optional
class CustomerServiceOrchestrator:
def __init__(
self,
client: OpenAI,
ticket_api: TicketAPI,
session_store: SessionStore,
knowledge_agent: KnowledgeRetrievalAgent
):
self.client = client
self.classifier = IntentClassifier(client)
self.billing_agent = BillingAgent(client, ticket_api)
self.technical_agent = TechnicalSupportAgent(client, knowledge_agent)
self.knowledge_agent = knowledge_agent
self.session_store = session_store
self.ticket_api = ticket_api
def process_message(
self,
customer_id: str,
message: str,
customer_context: Optional[Dict] = None
) -> Dict:
"""Process a single customer message through the full pipeline."""
# Load or create session
session = self.session_store.get_or_create_session(customer_id)
conversation_history = self.session_store.format_history(session["messages"])
# Step 1: Classify intent
classification = self.classifier.classify(
message=message,
conversation_history=conversation_history
)
# Store the customer message
self.session_store.append_message(
customer_id=customer_id,
role="customer",
content=message,
metadata={"classification": classification.model_dump()}
)
# Step 2: Check for immediate human escalation
if classification.requires_human:
return self._handle_escalation(
customer_id=customer_id,
reason=classification.reason_for_human or "Classifier flagged for human review",
summary=classification.summary,
classification=classification
)
# Step 3: Route to appropriate specialist
if classification.category == "billing":
result = self.billing_agent.handle(
customer_message=message,
customer_context=customer_context or {},
conversation_history=conversation_history
)
elif classification.category == "technical":
result = self.technical_agent.handle(
customer_message=message,
customer_context=customer_context or {},
conversation_history=conversation_history
)
elif classification.category in ["returns", "general_inquiry", "complaint"]:
# For these categories, use knowledge retrieval + general response
docs = self.knowledge_agent.search(query=message, top_k=5)
kb_result = self.knowledge_agent.generate_answer(
query=message,
retrieved_docs=docs
)
if kb_result["has_answer"]:
result = {
"action": "respond",
"response": kb_result["answer"],
"sources": kb_result["sources"],
"agent_type": "knowledge_agent"
}
else:
result = self._handle_escalation(
customer_id=customer_id,
reason="Knowledge base insufficient",
summary=classification.summary,
classification=classification
)
else:
result = self._handle_escalation(
customer_id=customer_id,
reason=f"Unsupported category: {classification.category}",
summary=classification.summary,
classification=classification
)
# Step 4: Check if specialist wants to escalate
if result.get("action") == "escalate":
return self._handle_escalation(
customer_id=customer_id,
reason=result.get("reason", "Specialist requested escalation"),
summary=result.get("summary", ""),
classification=classification,
specialist_context=result
)
# Step 5: Store the agent response
self.session_store.append_message(
customer_id=customer_id,
role="agent",
content=result.get("response", ""),
metadata={"agent_type": result.get("agent_type"), "action": result.get("action")}
)
return {
"customer_id": customer_id,
"response": result.get("response"),
"agent_type": result.get("agent_type"),
"classification": classification.model_dump(),
"sources": result.get("sources", []),
"requires_human": False,
"session_id": session["session_id"]
}
def _handle_escalation(
self,
customer_id: str,
reason: str,
summary: str,
classification: IntentClassification,
specialist_context: Optional[Dict] = None
) -> Dict:
"""Handle escalation to a human agent with full context preservation."""
session = self.session_store.get_or_create_session(customer_id)
# Create a ticket in the human queue
ticket = self.ticket_api.create_escalation_ticket(
customer_id=customer_id,
priority=classification.urgency,
category=classification.category,
summary=summary,
reason=reason,
conversation_transcript=self.session_store.format_history(session["messages"]),
classification=classification.model_dump(),
specialist_context=specialist_context
)
# Store escalation in session
self.session_store.append_message(
customer_id=customer_id,
role="system",
content=f"Escalated to human agent. Ticket #{ticket['ticket_id']} created.",
metadata={"ticket_id": ticket["ticket_id"], "priority": classification.urgency}
)
# Generate a customer-facing message
wait_time_estimate = {
"low": "within 24 hours",
"medium": "within 4 hours",
"high": "within 1 hour",
"critical": "within 15 minutes"
}
customer_message = (
f"I understand this requires personal attention. "
f"I've created ticket #{ticket['ticket_id']} and flagged it as {classification.urgency} priority. "
f"A human agent will review your case {wait_time_estimate.get(classification.urgency, 'soon')}. "
f"Here's a summary I've prepared for them: {summary}"
)
self.session_store.append_message(
customer_id=customer_id,
role="agent",
content=customer_message,
metadata={"action": "escalated"}
)
return {
"customer_id": customer_id,
"response": customer_message,
"agent_type": "escalation_handler",
"classification": classification.model_dump(),
"requires_human": True,
"ticket_id": ticket["ticket_id"],
"session_id": session["session_id"]
}
Building the Session Memory Store
The session store maintains conversation state. For production, use Redis or a database; here's a clean in-memory implementation suitable for development:
# memory/session_store.py
import uuid
from datetime import datetime, timedelta
from typing import Dict, List, Optional
class SessionStore:
def __init__(self, ttl_minutes: int = 60):
self.sessions: Dict[str, Dict] = {}
self.ttl = timedelta(minutes=ttl_minutes)
def get_or_create_session(self, customer_id: str) -> Dict:
"""Get existing session or create a new one."""
# Clean expired sessions
self._cleanup_expired()
# Check for existing active session
for session_id, session in self.sessions.items():
if session["customer_id"] == customer_id and not session.get("expired"):
# Update last activity
session["last_activity"] = datetime.now()
return session
# Create new session
session_id = f"session_{uuid.uuid4().hex[:12]}"
session = {
"session_id": session_id,
"customer_id": customer_id,
"created_at": datetime.now(),
"last_activity": datetime.now(),
"messages": [],
"context": {},
"expired": False
}
self.sessions[session_id] = session
return session
def append_message(self, customer_id: str, role: str, content: str, metadata: Optional[Dict] = None):
"""Add a message to the customer's active session."""
session = self.get_or_create_session(customer_id)
session["messages"].append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat(),
"metadata": metadata or {}
})
session["last_activity"] = datetime.now()
def format_history(self, messages: List[Dict], max_messages: int = 20) -> str:
"""Format message history for inclusion in LLM prompts."""
recent = messages[-max_messages:]
formatted = []
for msg in recent:
role_label = msg["role"].capitalize()
formatted.append(f"[{role_label}]: {msg['content']}")
return "\n".join(formatted)
def expire_session(self, session_id: str):
"""Mark a session as expired."""
if session_id in self.sessions:
self.sessions[session_id]["expired"] = True
def _cleanup_expired(self):
"""Remove sessions past their TTL."""
now = datetime.now()
expired_ids = []
for session_id, session in self.sessions.items():
if now - session["last_activity"] > self.ttl:
expired_ids.append(session_id)
for session_id in expired_ids:
del self.sessions[session_id]
Exposing the API
Finally, wrap everything in a FastAPI application