Building a Customer Service AI Agent That Executes Workflows (Not Just Answers Questions) - Full Architecture

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. The Core Architecture Shift 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: # Step 1: Classify customer intent intent = await classify_intent( message=conversation.latest_message, history=conversation.history ) # Step 2: Retrieve customer context from backend systems context = await retrieve_context( customer_id=conversation.customer_id, intent=intent ) # Step 3: Plan actions based on intent + context action_plan = await plan_actions( intent=intent, context=context, policy=load_policy(intent.type) ) # Step 4: Execute tools if the intent requires action tool_results = {} if action_plan.requires_tools: tool_results = await execute_tools(action_plan.tools) # Check escalation conditions before proceeding 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) ) # Step 5: Generate grounded response from results response = await generate_response( intent=intent, context=context, tool_results=tool_results ) # Step 6: Persist updated conversation state 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. Intent Classification 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": , "confidence": , "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. The Tool-Use Layer 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. Context Persistence Across Channels 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 information (survives channel switches) verified_identity: bool = False verified_order_id: Optional[str] = None verified_fields: Dict[str, Any] = field(default_factory=dict) # Resolution tracking attempted_resolutions: List[Dict] = field(default_factory=list) current_intent: Optional[str] = None # Escalation state escalation_reason: Optional[str] = None escalation_priority: str = "standard" # "standard" | "high" | "critical" # Cross-channel history 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: # Check for existing unresolved context across channels 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 # Load customer history to pre-populate 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 Routing 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: # Hard rules, always escalate regardless of confidence 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 # VIP / high-value customer handling if context.customer_tier == "enterprise": if len(context.attempted_resolutions) >= 2: return True # Confidence-based escalation if intent.confidence 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. Human Handoff Design 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: # Generate AI summary of conversation for the human agent 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