Agentic Tool Design.

# AUDIT: count + classify tools
SUPPORT_TOOLS = ["verify_customer", "lookup_order", "process_refund",
                 "escalate_to_human", "audit_log"]  # 5. At the optimum
ADMIN_TOOLS = ["create_user", "delete_user", "reset_password",
               "lock_account", "unlock_account", "audit_admin"]  # 6. Split

# WRONG: cram all 11 into one agent
# tools = SUPPORT_TOOLS + ADMIN_TOOLS  # 11. Routing accuracy drops ~32%

# RIGHT: two specialist agents, triage routes between them
def triage(user_request: str) -> str:
    """Tiny classifier. Pick the specialist agent."""
    if any(w in user_request.lower() for w in ["refund", "order", "ticket"]):
        return "support"
    if any(w in user_request.lower() for w in ["password", "account", "user"]):
        return "admin"
    return "support"  # default

def route(user_request: str) -> dict:
    specialist = triage(user_request)
    tools = SUPPORT_TOOLS if specialist == "support" else ADMIN_TOOLS
    return run_agent(tools=tools, message=user_request)

# Anthropic 4-line pattern. What / when / edge cases / ordering
TOOLS = [
    {
        "name": "verify_customer",
        "description": (
            "Look up a customer by customer_id and confirm they are active.\n"
            "Use this BEFORE any other tool that mentions the customer.\n"
            "Edge cases: returns 'not_found' if customer_id is missing or stale.\n"
            "Always run before lookup_order or process_refund."
        ),
        "input_schema": {
            "type": "object",
            "properties": {"customer_id": {"type": "string", "pattern": "^cust_[0-9]+$"}},
            "required": ["customer_id"],
        },
    },
    {
        "name": "process_refund",
        "description": (
            "Issue a refund to a verified customer up to the policy cap.\n"
            "Use ONLY after verify_customer has confirmed the customer is active.\n"
            "Edge cases: returns Permission error if amount > policy_cap (handled by hook).\n"
            "Never call before verify_customer; never call twice in one conversation."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "customer_id": {"type": "string"},
                "amount": {"type": "number", "minimum": 0},
                "reason": {"type": "string", "enum": ["damage", "wrong_item", "late", "other"]},
            },
            "required": ["customer_id", "amount", "reason"],
        },
    },
]

# .claude/hooks/refund_policy.py
import sys, json, os

POLICY_CAP = float(os.environ.get("REFUND_CAP", "500"))

def main():
    payload = json.loads(sys.stdin.read())
    tool_name = payload["tool_name"]
    tool_input = payload["tool_input"]

    if tool_name != "process_refund":
        sys.exit(0)  # not our concern, allow

    amount = tool_input.get("amount", 0)
    if amount > POLICY_CAP:
        # Stderr is fed back to the model as a tool_result with is_error=true
        print(
            f"refund ${amount} exceeds policy cap ${POLICY_CAP}; "
            f"escalate via escalate_to_human or reduce the amount",
            file=sys.stderr,
        )
        sys.exit(2)  # DENY

    # Additional structural checks. Verify customer is active, etc.
    if not tool_input.get("customer_id", "").startswith("cust_"):
        print("customer_id missing or malformed; call verify_customer first",
              file=sys.stderr)
        sys.exit(2)

    sys.exit(0)  # allow

if __name__ == "__main__":
    main()

# .claude/hooks/postuse_normalize.py
import sys, json, datetime

def normalize(tool_name: str, raw: dict) -> dict:
    """Project tool-specific raw output into a stable shape."""
    if tool_name == "lookup_order":
        return {
            "order_id": raw.get("id") or raw.get("order_id"),
            "status": (raw.get("status") or "unknown").upper(),
            "created_at": (
                datetime.datetime
                .fromtimestamp(raw["created_unix"])
                .isoformat() + "Z"
            ) if "created_unix" in raw else raw.get("created_at"),
            "total_cents": int(raw.get("total_cents", round(raw.get("total_dollars", 0) * 100))),
        }
    if tool_name == "verify_customer":
        return {
            "customer_id": raw.get("customer_id") or raw.get("id"),
            "active": bool(raw.get("active") or raw.get("is_active")),
            "tier": raw.get("tier") or raw.get("plan") or "standard",
        }
    return raw  # tools without a known shape pass through

def main():
    payload = json.loads(sys.stdin.read())
    tool_name = payload["tool_name"]
    raw_result = payload["tool_result"]

    normalized = normalize(tool_name, raw_result)
    payload["tool_result"] = normalized

    # Append canonical audit row
    with open("audit.jsonl", "a") as f:
        f.write(json.dumps({
            "ts": datetime.datetime.utcnow().isoformat() + "Z",
            "tool": tool_name,
            "input": payload["tool_input"],
            "output": normalized,
            "latency_ms": payload.get("latency_ms"),
        }) + "\n")

    # Pipe normalized payload back to stdout. SDK uses this as the new tool_result
    print(json.dumps(payload))
    sys.exit(0)

if __name__ == "__main__":
    main()

from enum import Enum
from typing import TypedDict

class ErrorBucket(str, Enum):
    TRANSIENT  = "Transient"
    PERMISSION = "Permission"
    DATA       = "Data"
    BUSINESS   = "Business"

class ToolError(TypedDict):
    bucket: ErrorBucket
    code: str          # e.g. "RATE_LIMITED", "FORBIDDEN", "INVALID_INPUT", "POLICY_BREACH"
    detail: str        # human-readable; for the model
    retryable: bool

def classify(http_status: int, body: dict) -> ToolError:
    """Project arbitrary backend error → 4-bucket contract."""
    if http_status >= 500 or http_status == 429:
        return {"bucket": ErrorBucket.TRANSIENT, "code": "RETRY",
                "detail": f"upstream {http_status}; retry with backoff",
                "retryable": True}
    if http_status in (401, 403):
        return {"bucket": ErrorBucket.PERMISSION, "code": "FORBIDDEN",
                "detail": "agent lacks permission; escalate, do not retry",
                "retryable": False}
    if http_status == 400:
        return {"bucket": ErrorBucket.DATA, "code": "INVALID_INPUT",
                "detail": body.get("message", "input failed validation"),
                "retryable": False}  # retry won't fix; user must reformulate
    if http_status == 422:
        return {"bucket": ErrorBucket.BUSINESS, "code": "POLICY_BREACH",
                "detail": body.get("message", "request violates business policy"),
                "retryable": False}
    return {"bucket": ErrorBucket.TRANSIENT, "code": "UNKNOWN",
            "detail": str(body), "retryable": True}

# Tool wrapper emits the contract via tool_result
def call_lookup_order(order_id: str) -> dict:
    resp = http_get(f"/orders/{order_id}")
    if resp.status_code != 200:
        return {"is_error": True, "error": classify(resp.status_code, resp.json())}
    return {"is_error": False, "data": resp.json()}

# auto. The right default for specialist agents
def support_agent(message: str):
    return client.messages.create(
        model="claude-sonnet-4.5",
        max_tokens=1024,
        tools=SUPPORT_TOOLS,
        tool_choice={"type": "auto"},  # agent decides
        messages=[{"role": "user", "content": message}],
    )

# forced. Only for mandatory extraction
def extract_one(email: str):
    return client.messages.create(
        model="claude-sonnet-4.5",
        max_tokens=1024,
        tools=[EXTRACT_TOOL],
        tool_choice={"type": "tool", "name": "extract_record"},  # MUST fire
        messages=[{"role": "user", "content": email}],
    )

# any. Rarely useful; "must call SOME tool but I won't pick which"
# def some_specialist():
#     return client.messages.create(
#         tool_choice={"type": "any"},
#         ...  # only for unusual flows where any of N tools is acceptable

# .claude/mcp.json. Declare the MCP servers your project uses
{
  "mcpServers": {
    "crm": {
      "command": "npx",
      "args": ["-y", "@yourorg/crm-mcp-server"],
      "env": {
        "CRM_API_KEY": "${CRM_API_KEY}",
        "CRM_BASE_URL": "https://crm.example.com"
      }
    }
  }
}

# In your agent. MCP tools auto-show up in the tools list
async def support_agent(message: str):
    # The CRM MCP server contributes verify_customer + lookup_order
    return await client.messages.create(
        model="claude-sonnet-4.5",
        max_tokens=1024,
        # tools are auto-loaded from MCP. Your local registry only adds:
        tools=[PROCESS_REFUND_TOOL, ESCALATE_TO_HUMAN_TOOL],
        tool_choice={"type": "auto"},
        messages=[{"role": "user", "content": message}],
        mcp_servers=["crm"],  # fictional pseudo-API; real call sites vary by SDK
    )

# eval/tool_routing.py. Measure first-call routing accuracy
INTENTS = [
    {"text": "I want a refund for order 12345",
     "expected_first_tool": "verify_customer"},
    {"text": "My order hasn't arrived",
     "expected_first_tool": "verify_customer"},
    {"text": "Cancel my account please",
     "expected_first_tool": "escalate_to_human"},
    # ... 47 more
]

def routing_accuracy() -> dict:
    correct = total = 0
    misses = []
    for intent in INTENTS:
        resp = client.messages.create(
            model="claude-sonnet-4.5",
            max_tokens=512,
            tools=SUPPORT_TOOLS,
            tool_choice={"type": "auto"},
            messages=[{"role": "user", "content": intent["text"]}],
        )
        first_tool = next(
            (b.name for b in resp.content if b.type == "tool_use"),
            None,
        )
        total += 1
        if first_tool == intent["expected_first_tool"]:
            correct += 1
        else:
            misses.append({
                "text": intent["text"],
                "expected": intent["expected_first_tool"],
                "got": first_tool,
            })
    return {
        "accuracy": correct / total,
        "n": total,
        "misses": misses[:10],  # first 10 for review
    }

# Re-run after every change to the tool registry; gate deploys on >= 95%