Add production-style support-agent Python example

Author: AgentPatterns

examples/examples/support-agent/python/README.md

@@ -0,0 +1,66 @@
+# Support Agent - Python Implementation
+
+Runnable support-agent example with safe-by-default behavior:
+- risk triage before drafting
+- no auto-send tool in allowlist
+- citation and policy checks
+- artifacts + audit trail
+- explicit human approval flag
+
+---
+
+## Quick start
+
+```bash
+python -m venv .venv && source .venv/bin/activate
+pip install -r requirements.txt
+
+# optional for low-risk draft flow
+export OPENAI_API_KEY="sk-..."
+export OPENAI_MODEL="gpt-4.1-mini"
+export OPENAI_TIMEOUT_SECONDS="60"
+
+# default ticket is high-risk (handoff path)
+python main.py
+
+# run low-risk drafting path
+TICKET_ID=T-1002 python main.py
+```
+
+## What this example demonstrates
+
+- Conservative triage (`billing_refund`, `security`, `legal`, `outage`)
+- Manual handoff for high-risk tickets
+- LLM draft only for non-high-risk tickets
+- Draft validation:
+  - no hard commitments
+  - required citations for policy-like claims
+- No write-side customer send action
+
+## Project layout
+
+```text
+python/
+  README.md
+  main.py
+  llm.py
+  gateway.py
+  policy.py
+  tools.py
+  requirements.txt
+```
+
+## Notes
+
+- This example uses in-memory stores for tickets/artifacts/audit.
+- Replace in-memory stores with real services in production.
+- Human approval is modeled via `requires_human_approval=true` output.
+
+## License
+
+MIT
+
+## Result schema
+
+- status is technical run state: success, blocked, stopped.
+- outcome is business result when status=success: handoff or draft_ready.

examples/examples/support-agent/python/gateway.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+from dataclasses import dataclass
+from typing import Any, Callable
+
+
+class StopRun(Exception):
+    def __init__(self, reason: str):
+        super().__init__(reason)
+        self.reason = reason
+
+
+@dataclass(frozen=True)
+class Budget:
+    max_tool_calls: int = 12
+    max_seconds: int = 30
+
+
+def _stable_json(value: Any) -> str:
+    if value is None or isinstance(value, (bool, int, float, str)):
+        return json.dumps(value, ensure_ascii=True, sort_keys=True)
+    if isinstance(value, list):
+        return "[" + ",".join(_stable_json(item) for item in value) + "]"
+    if isinstance(value, dict):
+        parts = []
+        for key in sorted(value):
+            parts.append(
+                json.dumps(str(key), ensure_ascii=True) + ":" + _stable_json(value[key])
+            )
+        return "{" + ",".join(parts) + "}"
+    return json.dumps(str(value), ensure_ascii=True)
+
+
+def args_hash(args: dict[str, Any]) -> str:
+    payload = _stable_json(args or {})
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]
+
+
+class ToolGateway:
+    def __init__(
+        self,
+        *,
+        allow: set[str],
+        registry: dict[str, Callable[..., dict[str, Any]]],
+        budget: Budget,
+    ):
+        self.allow = set(allow)
+        self.registry = registry
+        self.budget = budget
+        self.tool_calls = 0
+        self.started = time.monotonic()
+        self.seen_signatures: dict[str, int] = {}
+
+    def call(self, name: str, args: dict[str, Any]) -> dict[str, Any]:
+        elapsed = time.monotonic() - self.started
+        if elapsed > self.budget.max_seconds:
+            raise StopRun("max_seconds")
+
+        self.tool_calls += 1
+        if self.tool_calls > self.budget.max_tool_calls:
+            raise StopRun("max_tool_calls")
+
+        if name not in self.allow:
+            raise StopRun(f"tool_denied:{name}")
+
+        signature = f"{name}:{args_hash(args)}"
+        seen = self.seen_signatures.get(signature, 0) + 1
+        self.seen_signatures[signature] = seen
+        if seen > 2:
+            raise StopRun("loop_detected")
+
+        tool = self.registry.get(name)
+        if tool is None:
+            raise StopRun(f"tool_missing:{name}")
+
+        try:
+            result = tool(**args)
+        except TypeError as exc:
+            raise StopRun(f"tool_bad_args:{name}") from exc
+        except Exception as exc:
+            raise StopRun(f"tool_error:{name}") from exc
+
+        if isinstance(result, dict) and "error" in result:
+            raise StopRun(f"tool_result_error:{name}")
+
+        return result

examples/examples/support-agent/python/llm.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import json
+import os
+from typing import Any
+
+from openai import APIConnectionError, APITimeoutError, OpenAI
+
+MODEL = os.getenv("OPENAI_MODEL", "gpt-4.1-mini")
+LLM_TIMEOUT_SECONDS = float(os.getenv("OPENAI_TIMEOUT_SECONDS", "60"))
+
+
+class LLMTimeout(Exception):
+    pass
+
+
+SYSTEM_PROMPT = """
+You are a support drafting assistant.
+Return only one JSON object with exactly these keys:
+- customer_reply: string
+- internal_note: string
+- claims: array of {"kind": string, "text": string, "citation_id": string}
+- citations: array of {"id": string, "title": string}
+
+Rules:
+- Keep customer_reply concise and professional.
+- Do not promise refunds, credits, or guaranteed timelines.
+- If any policy claim is made, add citation_id.
+- Use only citation IDs present in provided policy/kb context.
+- Never include secrets.
+""".strip()
+
+
+def _get_client() -> OpenAI:
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        raise EnvironmentError(
+            "OPENAI_API_KEY is not set. Run: export OPENAI_API_KEY='sk-...'"
+        )
+    return OpenAI(api_key=api_key)
+
+
+def _as_claims(value: Any) -> list[dict[str, str]]:
+    claims: list[dict[str, str]] = []
+    if not isinstance(value, list):
+        return claims
+    for item in value:
+        if not isinstance(item, dict):
+            continue
+        claim = {
+            "kind": str(item.get("kind", "")).strip().lower(),
+            "text": str(item.get("text", "")).strip(),
+            "citation_id": str(item.get("citation_id", "")).strip(),
+        }
+        claims.append(claim)
+    return claims
+
+
+def _as_citations(value: Any) -> list[dict[str, str]]:
+    citations: list[dict[str, str]] = []
+    if not isinstance(value, list):
+        return citations
+    for item in value:
+        if not isinstance(item, dict):
+            continue
+        citation = {
+            "id": str(item.get("id", "")).strip(),
+            "title": str(item.get("title", "")).strip(),
+        }
+        if citation["id"]:
+            citations.append(citation)
+    return citations
+
+
+def generate_support_draft(
+    *,
+    ticket: dict[str, Any],
+    customer: dict[str, Any],
+    kb_matches: list[dict[str, Any]],
+    policy_matches: list[dict[str, Any]],
+) -> dict[str, Any]:
+    payload = {
+        "ticket": ticket,
+        "customer": customer,
+        "kb_matches": kb_matches,
+        "policy_matches": policy_matches,
+        "output_language": ticket.get("language", "en"),
+    }
+
+    client = _get_client()
+    try:
+        completion = client.chat.completions.create(
+            model=MODEL,
+            temperature=0,
+            timeout=LLM_TIMEOUT_SECONDS,
+            response_format={"type": "json_object"},
+            messages=[
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "user", "content": json.dumps(payload, ensure_ascii=True)},
+            ],
+        )
+    except (APITimeoutError, APIConnectionError) as exc:
+        raise LLMTimeout("llm_timeout") from exc
+
+    text = completion.choices[0].message.content or "{}"
+    try:
+        data = json.loads(text)
+    except json.JSONDecodeError:
+        data = {}
+
+    customer_reply = str(data.get("customer_reply", "")).strip()
+    internal_note = str(data.get("internal_note", "")).strip()
+    claims = _as_claims(data.get("claims", []))
+    citations = _as_citations(data.get("citations", []))
+
+    if not customer_reply:
+        customer_reply = (
+            "Thanks for your message. We are reviewing your request with the support team "
+            "and will follow up shortly."
+        )
+    if not internal_note:
+        internal_note = "Draft created. Please verify policy claims before approval."
+
+    return {
+        "customer_reply": customer_reply,
+        "internal_note": internal_note,
+        "claims": claims,
+        "citations": citations,
+    }