Capstone 7-B — Agent Evolution: Order Exception

mkdir -p agent-iter1-raw/mock_data && cd agent-iter1-raw
python -m venv venv && source venv/bin/activate    # Windows: venv\Scripts\activate
pip install "anthropic>=0.40" "fastapi>=0.110" "uvicorn>=0.27" "pydantic>=2.0"

10 POs

"""tools.py — order-exception tool schemas + executors."""
import json
from pathlib import Path

DATA = Path("mock_data")
ORDERS = json.loads((DATA / "orders.json").read_text())
TRACKING = json.loads((DATA / "tracking.json").read_text())
CONTRACTS = json.loads((DATA / "contracts.json").read_text())
INVENTORY = json.loads((DATA / "inventory.json").read_text())
CUSTOMERS = json.loads((DATA / "customers.json").read_text())

TOOLS = [
    {
        "name": "get_order_details",
        "description": "Get full PO details: line items, status, dates, tracking numbers, customer.",
        "input_schema": {
            "type": "object",
            "properties": {"po_number": {"type": "string", "pattern": "^PO-[0-9]{4}-[0-9]{4}$"}},
            "required": ["po_number"],
        },
    },
    {
        "name": "track_shipment",
        "description": "Get carrier status for a tracking number (UPS/FedEx/DHL).",
        "input_schema": {
            "type": "object",
            "properties": {"tracking_number": {"type": "string"}},
            "required": ["tracking_number"],
        },
    },
    {
        "name": "check_contract_pricing",
        "description": "Return contract unit price for a customer + SKU. Used to detect discrepancies.",
        "input_schema": {
            "type": "object",
            "properties": {"customer_id": {"type": "string"}, "sku": {"type": "string"}},
            "required": ["customer_id", "sku"],
        },
    },
    {
        "name": "query_inventory",
        "description": "Return stock available for a SKU at a specific warehouse.",
        "input_schema": {
            "type": "object",
            "properties": {"sku": {"type": "string"}, "warehouse": {"type": "string"}},
            "required": ["sku", "warehouse"],
        },
    },
    {
        "name": "draft_notification",
        "description": "Generate an exception-notification email draft, tier-aware tone.",
        "input_schema": {
            "type": "object",
            "properties": {
                "customer_id": {"type": "string"},
                "exception_type": {"type": "string",
                                   "enum": ["DELAYED", "PARTIAL", "PRICING", "QUALITY_HOLD"]},
                "resolution": {"type": "string"},
            },
            "required": ["customer_id", "exception_type", "resolution"],
        },
    },
]

def execute_tool(name, args):
    if name == "get_order_details":
        return ORDERS.get(args["po_number"], {"error": "PO not found"})
    if name == "track_shipment":
        return TRACKING.get(args["tracking_number"], {"error": "tracking not found"})
    if name == "check_contract_pricing":
        key = f"{args['customer_id']}_{args['sku']}"
        return CONTRACTS.get(key, {"error": "no contract on file"})
    if name == "query_inventory":
        sku = INVENTORY.get(args["sku"], {})
        return sku.get("warehouses", {}).get(args["warehouse"],
                                             {"error": "no inventory record"})
    if name == "draft_notification":
        cust = CUSTOMERS.get(args["customer_id"], {})
        tier = cust.get("tier", "SMB")
        # Tier-aware salutation/tone
        salutation = "Team," if tier == "ENTERPRISE" else f"Hi {cust.get('name','team')},"
        sign_off = ("We will follow up before SLA expiry. — Operations"
                    if tier == "ENTERPRISE"
                    else "Let us know how we can help! — Customer Success")
        body = (f"Regarding {args['exception_type']} on your recent order:\n\n"
                f"{args['resolution']}\n\n{sign_off}")
        return {"to": cust.get("csm_email", "ops@example.com"),
                "tier": tier, "salutation": salutation, "body": body}
    raise ValueError(f"Unknown tool: {name}")

"""agent.py — the raw API loop. Everything is hand-coded."""
import json, anthropic
from tools import TOOLS, execute_tool

client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"
SYSTEM = """You are an order-exception specialist. For every flagged PO:
1. get_order_details to understand the situation
2. track_shipment for any tracking numbers (CARRIER VIEW)
3. check_contract_pricing for any line item with potential discrepancy
4. query_inventory for any unshipped or backordered SKUs
5. draft_notification with the right exception_type and a clear resolution

ALWAYS investigate before drafting. Never recommend a credit without checking
contract pricing AND inventory. Match the customer tier's tone (terse for
ENTERPRISE, warmer for SMB). Never overpromise — cite SLA hours, not specifics."""

MAX_TURNS = 12

def _run_messages(messages: list) -> tuple[str, list]:
    """Drive the loop on the given messages list. Returns (final_text, messages).
    Used by run_agent() for single-shot queries and by session.chat() for multi-turn."""
    for turn in range(MAX_TURNS):
        response = client.messages.create(
            model=MODEL, max_tokens=4096, system=SYSTEM,
            tools=TOOLS, messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            text = next(b.text for b in response.content if b.type == "text")
            return text, messages

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    try:
                        result = execute_tool(block.name, block.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result, default=str),
                        })
                    except Exception as e:
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": f"ERROR: {e}",
                            "is_error": True,
                        })
            messages.append({"role": "user", "content": tool_results})
            continue

        raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason}")
    raise RuntimeError(f"Agent exceeded {MAX_TURNS} turns without finishing")

def run_agent(question: str) -> str:
    """Single-shot entry point. Wraps _run_messages with a fresh history."""
    text, _ = _run_messages([{"role": "user", "content": question}])
    return text

if __name__ == "__main__":
    q = "PO-2024-5678 from Acme Corp has a flagged exception. Investigate, " \
        "determine the root cause, and draft a customer notification."
    print(run_agent(q))

"""guardrails.py — B2B-aware checks for order agents."""
import re

PO_RE = re.compile(r"^PO-[0-9]{4}-[0-9]{4}$")
SKU_RE = re.compile(r"^[A-Z]{3}-[0-9]{4}$")
COST_LIMIT_TOKENS = 50_000
CIRCUIT_FAIL_THRESHOLD = 3

# Tokens that should NOT appear in ENTERPRISE-tier drafts
ENTERPRISE_BANNED_TOKENS = ["hey there", "hi there", "no worries", "super sorry"]
# Tokens that should NOT appear in any draft (overpromise)
OVERPROMISE_TOKENS = ["guarantee", "definitely tomorrow", "100% certain"]

def validate_input(tool_name: str, args: dict):
    if tool_name == "get_order_details":
        if not PO_RE.match(args.get("po_number", "")):
            raise ValueError(f"Invalid PO format: {args.get('po_number')}")
    if tool_name in ("check_contract_pricing", "query_inventory"):
        if not SKU_RE.match(args.get("sku", "")):
            raise ValueError(f"Invalid SKU format: {args.get('sku')}")

def check_draft_tone(draft: dict):
    """Run AFTER draft_notification. Raise if tone mismatch or overpromise."""
    body = draft.get("body", "").lower()
    tier = draft.get("tier", "SMB")
    if tier == "ENTERPRISE":
        for tok in ENTERPRISE_BANNED_TOKENS:
            if tok in body:
                raise ValueError(f"Tone mismatch: '{tok}' in ENTERPRISE draft")
    for tok in OVERPROMISE_TOKENS:
        if tok in body:
            raise ValueError(f"Overpromise detected: '{tok}'")

class CircuitBreaker:
    def __init__(self): self.failures = 0
    def record(self, ok):
        self.failures = 0 if ok else self.failures + 1
        if self.failures >= CIRCUIT_FAIL_THRESHOLD:
            raise RuntimeError("Circuit breaker tripped — aborting")

# Add to imports at the top of agent.py:
from guardrails import (validate_input, check_draft_tone,
                        CircuitBreaker, COST_LIMIT_TOKENS)   # NEW

_breaker = CircuitBreaker()        # NEW — module-level instance

def _run_messages(messages: list) -> tuple[str, list]:
    total_tokens = 0                # NEW — cost cap counter
    for turn in range(MAX_TURNS):
        response = client.messages.create(
            model=MODEL, max_tokens=4096, system=SYSTEM,
            tools=TOOLS, messages=messages,
        )
        total_tokens += (response.usage.input_tokens
                         + response.usage.output_tokens)              # NEW
        if total_tokens > COST_LIMIT_TOKENS:                          # NEW
            raise RuntimeError(f"Cost cap exceeded: {total_tokens} tokens")

        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            text = next(b.text for b in response.content if b.type == "text")
            return text, messages

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    try:
                        validate_input(block.name, block.input)        # NEW — PO/SKU check
                        result = execute_tool(block.name, block.input)
                        # B2B-specific: tone-check draft_notification results.
                        if block.name == "draft_notification":         # NEW
                            check_draft_tone(result)                    # NEW — raises on bad tone
                        _breaker.record(ok=True)                        # NEW
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result, default=str),
                        })
                    except Exception as e:
                        _breaker.record(ok=False)                       # NEW — trips on 3rd fail
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": f"ERROR: {e}",
                            "is_error": True,
                        })
            messages.append({"role": "user", "content": tool_results})
            continue

        raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason}")

    raise RuntimeError(f"Agent exceeded {MAX_TURNS} turns without finishing")

# Temporarily edit your draft_notification mock in tools.py to inject bad tone:
#   "Hi there team! No worries about PO-2024-5678 ..."
# Run the agent. The check_draft_tone hook should raise ValueError,
# the error flows back as is_error: true, and Claude redrafts with formal tone.
python agent.py

"""audit.py — SOX-compliant audit log for B2B order operations."""
import json, datetime
def append_audit(tool_name, args, result, tokens, po_number):
    rec = {
        "ts": datetime.datetime.utcnow().isoformat() + "Z",
        "po_number": po_number,
        "tool": tool_name,
        "input": args,
        "output_summary": str(result)[:200],
        "tokens": tokens,
    }
    with open("audit_log.jsonl", "a") as f:
        f.write(json.dumps(rec) + "\n")

# Add to imports at the top of agent.py:
from audit import append_audit                    # NEW

# Inside _run_messages, after `result = execute_tool(...)` and the tone check:
                        if block.name == "draft_notification":
                            check_draft_tone(result)
                        _breaker.record(ok=True)
                        # Extract PO number from the user's first message for audit context
                        po = next(
                            (w for w in messages[0]["content"].split() if w.startswith("PO-")),
                            "default"
                        )
                        append_audit(                              # NEW
                            tool_name=block.name,
                            args=block.input,
                            result=result,
                            tokens=total_tokens,
                            po_number=po,
                        )
                        tool_results.append({...})
                        # ... rest of try-block unchanged ...

cat audit_log.jsonl   # macOS/Linux
type audit_log.jsonl  # Windows cmd

"""session.py — multi-turn PO sessions over the same _run_messages helper."""
from agent import _run_messages

SESSIONS: dict[str, list] = {}   # po_number -> messages list
WINDOW = 24                      # 5 tools per turn × ~5 turns + buffer

def chat(po_number: str, user_msg: str) -> str:
    """Append the user's message to the PO session, run the loop, return the answer."""
    msgs = SESSIONS.setdefault(po_number, [])
    msgs.append({"role": "user", "content": user_msg})
    answer, msgs = _run_messages(msgs)
    # Sliding window: keep only the last WINDOW messages.
    SESSIONS[po_number] = msgs[-WINDOW:]
    return answer

python -c "
from session import chat
print(chat('PO-2024-5678', 'PO-2024-5678 from Acme Corp has a flagged exception. Investigate and draft a notification.'))
print('---')
print(chat('PO-2024-5678', 'What would change if we expedite-ship from Newark instead of waiting for Memphis restock?'))
"

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent import run_agent
from session import chat

app = FastAPI()
class Q(BaseModel): question: str
class C(BaseModel): po_number: str; message: str

@app.get("/health")
def health(): return {"status": "ok"}

@app.post("/exception")
def exception_ep(q: Q):
    try: return {"report": run_agent(q.question)}
    except Exception as e: raise HTTPException(500, str(e))

@app.post("/chat")
def chat_ep(c: C):
    return {"answer": chat(c.po_number, c.message)}

anthropic>=0.40
fastapi>=0.110
uvicorn>=0.27
pydantic>=2.0

uvicorn server:app --reload --port 8000
# In another terminal:
curl localhost:8000/health
curl -X POST localhost:8000/exception -H "Content-Type: application/json" \
     -d '{"question":"PO-2024-5678 from Acme Corp has a flagged exception. Investigate."}'

docker build -t iter1-b .
docker run --rm -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY iter1-b

mkdir agent-iter2-sdk && cd agent-iter2-sdk
python -m venv venv && source venv/bin/activate    # Windows: venv\Scripts\activate
pip install "claude-agent-sdk>=0.2" "fastapi>=0.110" "uvicorn>=0.27" "pydantic>=2.0"
npm i -g @anthropic-ai/claude-code   # if not already installed
claude
> /init

> Create agent.py using `claude-agent-sdk`. Define five order-ops tools as
> @tool-decorated async functions returning MCP-shaped {"content":[...]}:
> get_order_details, track_shipment, check_contract_pricing, query_inventory,
> draft_notification (tier-aware tone). Wire them into a create_sdk_mcp_server,
> build ClaudeAgentOptions with the system prompt from CLAUDE.md, and expose
> async run(question) driving query() and concatenating AssistantMessage text.

{
  "hooks": {
    "PreToolUse": [
      { "matcher": "*", "command": "python hooks/log_call.py" }
    ],
    "PostToolUse": [
      { "matcher": "mcp__ops__draft_notification",
        "command": "python hooks/tone_check.py" },
      { "matcher": "*", "command": "python hooks/audit.py" }
    ]
  }
}

"""hooks/log_call.py — PreToolUse: print to stderr, pass through stdout."""
import sys, json, datetime

payload = json.load(sys.stdin)
ts = datetime.datetime.utcnow().isoformat() + "Z"
print(f"[{ts}] PRE  {payload.get('tool_name','?')}({payload.get('tool_input',{})})",
      file=sys.stderr)
json.dump(payload, sys.stdout)   # pass-through

# tone_check should reject ENTERPRISE drafts with banned tokens
echo '{"tool_name":"mcp__ops__draft_notification","tool_result":{"content":[{"type":"text","text":"{\"body\":\"Hi there team!\",\"tier\":\"ENTERPRISE\"}"}]}}' \
  | python hooks/tone_check.py
# Should rewrite tool_result to an error payload prompting redraft.

# audit should append a record keyed by po_number
echo '{"tool_name":"get_order_details","tool_input":{"po_number":"PO-2024-5678"},"tool_result":"ok"}' \
  | python hooks/audit.py
cat audit_log.jsonl | tail -1

python -c "import asyncio, agent; print(asyncio.run(agent.run('Investigate PO-2024-5678 from Acme Corp')))"

"""sessions.py — multi-PO via SDK resume tokens."""
from dataclasses import replace
from claude_agent_sdk import query, AssistantMessage
from agent import OPTIONS

SESSIONS: dict[str, str] = {}   # po_number -> resume token

async def _drive(prompt, options):
    parts, sid = [], None
    async for msg in query(prompt=prompt, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if getattr(block, "text", None): parts.append(block.text)
        s = getattr(msg, "session_id", None)
        if s: sid = s
    return "\n".join(parts), sid

async def chat(po_number: str, msg: str) -> str:
    resume = SESSIONS.get(po_number)
    options = replace(OPTIONS, resume=resume) if resume else OPTIONS
    text, sid = await _drive(msg, options)
    if sid: SESSIONS[po_number] = sid
    return text

async def what_if(po_number: str, hypothetical: str) -> str:
    """Fork: 'what if we expedite-ship from Newark instead?' — do NOT save the new sid."""
    resume = SESSIONS.get(po_number)
    options = replace(OPTIONS, resume=resume) if resume else OPTIONS
    text, _ = await _drive(hypothetical, options)
    return text

python -c "
import asyncio
from sessions import chat, what_if

async def main():
    print('T1:', await chat('PO-2024-5678', 'PO-2024-5678 from Acme has a flagged exception. Investigate.'))
    print('FORK:', await what_if('PO-2024-5678', 'What if we expedite-ship from Newark instead of waiting for Memphis?'))
    print('T2:', await chat('PO-2024-5678', 'Stick with the original plan. Draft the customer notification.'))

asyncio.run(main())
"

---
description: Investigate an order exception by PO number
argument-hint: [po_number]
---
Look up $ARGUMENTS in mock_data/orders.json. Build the question string.
Run `python -c "import asyncio, agent; print(asyncio.run(agent.run(q)))"`
where q is the question. Print the exception report, draft, total tokens,
total cost from the ResultMessage emitted by query().

> Create server.py and Dockerfile. Endpoints: GET /health,
> POST /exception (single-shot), POST /chat (po_number + message).
> Async FastAPI handlers awaiting agent.run() and sessions.chat() (both
> async coroutines from claude-agent-sdk). python:3.11-slim base, install
> claude-agent-sdk + dependencies, expose 8000. Mount .claude/ into the
> container so settings.json + hook scripts resolve at runtime.

uvicorn server:app --reload --port 8000
# In another terminal:
curl localhost:8000/health
curl -X POST localhost:8000/exception -H "Content-Type: application/json" \
     -d '{"question":"Investigate PO-2024-5678"}'

# Then build the container:
docker build -t iter2-b .
docker run --rm -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY iter2-b