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": <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.

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 < 0.65:
        return True

    if tool_results.get("resolution_confidence", 1.0) < 0.70:
        return True

    # Time and turn limits
    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.

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_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
    }

    # Route to appropriate queue based on priority and type
    queue = determine_queue(reason, context.customer_tier)
    ticket_id = await helpdesk.create_escalation(queue, handoff_payload)

    # Inform the customer honest about what's happening
    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.

What This Architecture Doesn't Solve

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, [e-mail]