Most customer service AI implementations answer questions. They retrieve relevant information from a knowledge base, synthesize a response, and hand the conversation back to the customer.
That's a chatbot. A good one, in 2026, but still a chatbot.
A customer service agent that executes workflows does something different. It processes the refund. It updates the account. It triggers the return label. It does what the customer asked, inside the systems that matter, and sends confirmation when it's done. The difference isn't the model, it's the architecture around the model, specifically the tool-use layer and how everything connecting to it is designed.
This is the architecture we use for production customer service agents. It's the part that most tutorials skip.
A Q&A agent has one primary operation: retrieve context, generate response. The loop is simple.
A workflow-execution agent has three: classify intent, execute tools, generate response. The middle step is where production complexity lives.
Here's the agent loop that governs everything:
from anthropic import AsyncAnthropic
from typing import Optional
import asyncio
client = AsyncAnthropic()
async def agent_loop(conversation: Conversation) -> AgentResponse:
intent = await classify_intent(
message=conversation.latest_message,
history=conversation.history
)
context = await retrieve_context(
customer_id=conversation.customer_id,
intent=intent
)
action_plan = await plan_actions(
intent=intent,
context=context,
policy=load_policy(intent.type)
)
tool_results = {}
if action_plan.requires_tools:
tool_results = await execute_tools(action_plan.tools)
if should_escalate(tool_results, intent, context):
return await escalate_to_human(
conversation=conversation,
context=context,
tool_results=tool_results,
reason=determine_escalation_reason(intent, tool_results)
)
response = await generate_response(
intent=intent,
context=context,
tool_results=tool_results
)
await persist_context(conversation, response, tool_results)
return response
The key difference from a retrieval-only agent: step 4 executes real operations against real systems. The agent isn't describing what should happen. It's making it happen.
Before any tool call happens, the agent needs to know what category of request it's dealing with. Intent classification determines which tools get considered and which policy rules apply.
INTENT_CATEGORIES = [
"order_status",
"return_request",
"refund_request",
"account_update",
"billing_dispute",
"product_question",
"complaint",
"explicit_escalation"
]
async def classify_intent(message: str, history: list) -> Intent:
response = await client.messages.create(
model="claude-sonnet-4-5",
max_tokens=256,
system="""Classify the customer message into exactly one intent category.
Return JSON: {"type": <category>, "confidence": <0-1>, "entities": {}}
Categories: order_status, return_request, refund_request, account_update,
billing_dispute, product_question, complaint, explicit_escalation""",
messages=[
{"role": "user", "content": f"History: {history[-3:]}\nMessage: {message}"}
]
)
return Intent.from_json(response.content[0].text)
Confidence scoring here matters beyond routing. When the intent classification confidence falls below 0.70, that's a signal for tighter tool permission scoping and a lower escalation threshold, the agent is operating with more uncertainty about what the customer actually needs.
This is where the architecture diverges from a retrieval-only design. Tools are the interface between the agent and your backend systems, CRM, order management, payment processor, helpdesk.
CUSTOMER_SERVICE_TOOLS = [
{
"name": "lookup_order",
"description": "Retrieve current order status, tracking, and line items for a customer order",
"input_schema": {
"type": "object",
"properties": {
"order_id": {"type": "string"},
"customer_id": {"type": "string"}
},
"required": ["order_id", "customer_id"]
}
},
{
"name": "process_refund",
"description": "Initiate a refund for an eligible order within policy parameters",
"input_schema": {
"type": "object",
"properties": {
"order_id": {"type": "string"},
"refund_amount": {"type": "number"},
"reason": {"type": "string"},
"policy_check_passed": {"type": "boolean"}
},
"required": ["order_id", "refund_amount", "reason", "policy_check_passed"]
}
},
{
"name": "create_return_label",
"description": "Generate a prepaid return shipping label and initiate return workflow",
"input_schema": {
"type": "object",
"properties": {
"order_id": {"type": "string"},
"return_reason": {"type": "string"}
},
"required": ["order_id", "return_reason"]
}
},
{
"name": "update_account_field",
"description": "Update a customer account detail after identity verification",
"input_schema": {
"type": "object",
"properties": {
"customer_id": {"type": "string"},
"field": {"type": "string", "enum": ["email", "address", "phone"]},
"new_value": {"type": "string"},
"verified": {"type": "boolean"}
},
"required": ["customer_id", "field", "new_value", "verified"]
}
}
]
Notice policy_check_passed
and verified
as required fields in the refund and account update tools. The agent cannot call these tools without explicitly confirming that policy eligibility has been checked and identity has been verified. This is enforcement at the tool signature level, not at the prompt level, a much harder constraint to bypass.
A customer who starts a conversation in web chat, follows up by email, and then calls voice support should not have to re-explain their situation at each handoff. Context persistence is what makes multi-channel support feel like a single conversation rather than three separate ones.
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
@dataclass
class ConversationContext:
customer_id: str
conversation_id: str
channel: str # "web_chat" | "email" | "voice" | "whatsapp"
verified_identity: bool = False
verified_order_id: Optional[str] = None
verified_fields: Dict[str, Any] = field(default_factory=dict)
attempted_resolutions: List[Dict] = field(default_factory=list)
current_intent: Optional[str] = None
escalation_reason: Optional[str] = None
escalation_priority: str = "standard" # "standard" | "high" | "critical"
prior_conversations: List[str] = field(default_factory=list)
unresolved_issues: List[str] = field(default_factory=list)
async def load_or_create_context(customer_id: str, channel: str) -> ConversationContext:
existing = await redis_client.get(f"context:{customer_id}:active")
if existing:
context = ConversationContext.from_json(existing)
context.channel = channel # Update current channel
return context
customer = await crm.get_customer(customer_id)
return ConversationContext(
customer_id=customer_id,
conversation_id=generate_id(),
channel=channel,
prior_conversations=customer.recent_conversation_ids,
unresolved_issues=customer.open_tickets
)
The verified_fields
dictionary matters specifically. When a customer verifies their identity in the web chat session, that verification persists into the voice session. The agent in the new channel knows what's already been confirmed and doesn't ask the customer to reverify.
Escalation logic is where production quality separates from demo quality. An agent that escalates incorrectly frustrates customers. An agent that fails to escalate when it should creates liability.
LEGAL_KEYWORDS = [
"attorney", "lawyer", "lawsuit", "legal action",
"sue", "court", "fraud", "chargeback dispute"
]
SECURITY_KEYWORDS = [
"hacked", "unauthorized", "data breach",
"identity theft", "fraud", "compromised"
]
def should_escalate(
tool_results: Dict,
intent: Intent,
context: ConversationContext
) -> bool:
if intent.type == "explicit_escalation":
return True
if contains_any(context.latest_message, LEGAL_KEYWORDS):
return True
if contains_any(context.latest_message, SECURITY_KEYWORDS):
return True
if context.customer_tier == "enterprise":
if len(context.attempted_resolutions) >= 2:
return True
if intent.confidence < 0.65:
return True
if tool_results.get("resolution_confidence", 1.0) < 0.70:
return True
if context.turn_count > 8:
return True
if context.elapsed_seconds > 600:
return True
return False
The tiered escalation threshold for enterprise customers (>= 2 attempts
vs the standard > 8 turns
) reflects a business decision that high-value customers get faster human access. That policy lives in code, not in a prompt, which means it's enforceable and auditable.
The handoff payload is the most underbuilt piece of most agent architectures. If the human agent who picks up the escalation has to re-read the transcript and reconstruct context from scratch, the escalation experience is worse than if the customer had just called a human from the start.
async def escalate_to_human(
conversation: Conversation,
context: ConversationContext,
tool_results: Dict,
reason: str
) -> AgentResponse:
summary = await generate_handoff_summary(conversation, context, tool_results)
handoff_payload = {
"conversation_id": conversation.id,
"customer": {
"id": context.customer_id,
"name": context.customer_name,
"tier": context.customer_tier,
"lifetime_value": context.customer_ltv,
"sentiment": context.sentiment_score
},
"summary": summary,
"escalation_reason": reason,
"escalation_priority": context.escalation_priority,
"attempted_resolutions": context.attempted_resolutions,
"verified_context": context.verified_fields,
"recommended_action": await suggest_next_action(context, tool_results),
"full_transcript": conversation.messages,
"open_tickets": context.unresolved_issues
}
queue = determine_queue(reason, context.customer_tier)
ticket_id = await helpdesk.create_escalation(queue, handoff_payload)
return AgentResponse(
message=f"I'm connecting you with a specialist who can help with this. "
f"They'll have the full context of our conversation "
f"you won't need to repeat anything. Reference: {ticket_id}",
action="escalate",
ticket_id=ticket_id
)
The recommended_action
field in the payload is the piece that changes average handle time most significantly. When the human agent opens the case, they get a structured recommendation, not a transcript to reconstruct. The AI hasn't transferred the problem, it has transferred a decision-ready package.
Building this correctly takes the agent from Q&A to workflow execution. What it doesn't solve is the integration layer beneath the tools, getting process_refund()
to connect reliably to your actual payment processor, with your actual permission model, handling your actual error states.
The tool signatures above are intentionally clean. The implementations behind them are where the production complexity lives: rate limiting, retry logic, circuit breakers for when downstream APIs fail mid-conversation, audit logging for every write operation.
That gap between a tool defined and a tool that works under production load is where most enterprise agent projects either invest properly or discover they should have.
The chat is the easy part. The workflow execution and backend integration is where production agents are made or broken. We wrote the full build guide covering the complete eight-layer architecture, reliability patterns, cost modeling, and the failure modes that break agents at scale.
How to Build a 24/7 AI Customer Service Agent, Enterprise Guide
Dextra Labs builds production AI agent systems for enterprise customer service, finance, and operations. If your agent architecture is at the integration layer and you want a technical review, hello@dextralabs.com