Customer Support Agent with OpenClaw

# agents/support-agent.yaml
name: support-agent
model: gpt-4o
temperature: 0.3
tools:
  - knowledge_base_search   # semantic search over your docs
  - ticket_api              # create/update CRM tickets
  - customer_lookup         # read customer profile (read-only)
knowledge_base:
  source: "./kb/faq-articles/"
  chunk_size: 512
  top_k: 5
system: |
  You are a friendly customer support agent for OpenClaw AI.
  Always search the knowledge base before answering.
  If confidence is below 0.7, say you are not sure and offer to escalate.
  Never invent product features that don't exist.
  Keep replies concise — under 150 words unless a step-by-step guide is needed.
  Tone: professional but warm.

from openclaw import Agent

agent = Agent.from_config("agents/support-agent.yaml")

def answer_ticket(ticket_text: str, customer_id: str) -> dict:
    response = agent.run(
        f"Customer ID: {customer_id}\n"
        f"Question: {ticket_text}\n\n"
        "Search the knowledge base and answer. "
        "Return JSON: {\"reply\": \"...\", \"confidence\": 0.0-1.0, "
        "\"kb_article_ids\": [], \"needs_escalation\": false}"
    )
    import json
    result = json.loads(response)
    if result["confidence"] < 0.7 or result["needs_escalation"]:
        result["reply"] += (
            "\n\nI'll connect you with one of our specialists who "
            "can give you a definitive answer. Expect a reply within 2 hours."
        )
        result["needs_escalation"] = True
    return result

import json

CATEGORIES = ["billing", "technical", "account", "feature_request", "abuse", "other"]
PRIORITIES = ["low", "normal", "high", "urgent"]

def classify_ticket(text: str) -> dict:
    result = agent.run(
        f"Classify this support ticket:\n\n{text}\n\n"
        f"Return JSON: {{\"category\": one of {CATEGORIES}, "
        f"\"priority\": one of {PRIORITIES}, "
        "\"sentiment\": \"positive|neutral|negative|angry\", "
        "\"summary\": \"one sentence\"}}"
    )
    return json.loads(result)

def triage_ticket(ticket: dict) -> dict:
    """Auto-escalate angry or urgent tickets."""
    classification = classify_ticket(ticket["body"])
    
    if classification["sentiment"] == "angry" or classification["priority"] == "urgent":
        # Immediate escalation — no auto-reply
        return {
            "action": "escalate",
            "team_queue": "tier-2-support",
            "reason": f"Sentiment: {classification['sentiment']}, "
                      f"Priority: {classification['priority']}",
            "classification": classification,
        }
    
    answer = answer_ticket(ticket["body"], ticket["customer_id"])
    return {"action": "auto_reply", "reply": answer["reply"], "classification": classification}

import os, requests

ZENDESK_BASE = os.environ["ZENDESK_BASE_URL"]  # https://yourcompany.zendesk.com
ZENDESK_TOKEN = os.environ["ZENDESK_API_TOKEN"]
ZENDESK_EMAIL = os.environ["ZENDESK_EMAIL"]

def update_zendesk_ticket(ticket_id: int, reply: str, status: str = "pending") -> bool:
    payload = {
        "ticket": {
            "status": status,
            "comment": {"body": reply, "public": True},
        }
    }
    resp = requests.put(
        f"{ZENDESK_BASE}/api/v2/tickets/{ticket_id}.json",
        json=payload,
        auth=(f"{ZENDESK_EMAIL}/token", ZENDESK_TOKEN),
        timeout=10,
    )
    resp.raise_for_status()
    return True