AI Automation Code Examples: Triage For An Operations Inbox With Claude, Measured Before It Routes Anything

"""Operations inbox triage: classify, enrich and route. People act, the agent never replies.

The model classifies each email through one strict tool. Forced tool use (tool_choice "any" or "tool")
returns a 400 error on Claude Opus 5.5 and Sonnet 5.5, so this uses tool_choice "auto" and treats a
reply without the tool call as "needs a person". Requests to change settlement instructions always go to
manual verification by call-back, whatever the model says.

Requires anthropic 1.14.0. Runs against a stub client that returns real SDK types, so no API key is
needed. Tested on Python 3.14.4 and 3.11.14 with mypy --strict. Sample emails and references only.
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass
from email import message_from_string
from email.message import Message as EmailMessage

from anthropic.types import Message, ToolParam, ToolUseBlock

CATEGORIES = ["trade_query", "ssi_change", "document_request", "valuation_request", "other"]
CLASSIFY: ToolParam = {
    "name": "classify_email",
    "description": "Classify one operations email and extract references. Use null when absent.",
    "strict": True,
    "input_schema": {
        "type": "object",
        "properties": {
            "category": {"type": "string", "enum": CATEGORIES},
            "trade_ref": {"type": ["string", "null"]},
            "account_ref": {"type": ["string", "null"]},
        },
        "required": ["account_ref", "category", "trade_ref"],
        "additionalProperties": False,
    },
}
QUEUES = {"trade_query": "middle-office", "document_request": "client-services",
          "valuation_request": "client-services", "other": "ops-triage"}


@dataclass(frozen=True)
class Routed:
    subject: str
    category: str
    queue: str
    context: dict[str, str]


def classify(create: Callable[..., Message], email: EmailMessage, model: str) -> dict[str, object] | None:
    body = email.get_payload()
    reply = create(model=model, max_tokens=512, tools=[CLASSIFY], tool_choice={"type": "auto"},
                   messages=[{"role": "user", "content": f"Subject: {email['Subject']}\n\n{body}"}])
    calls = [b for b in reply.content if isinstance(b, ToolUseBlock) and b.name == "classify_email"]
    return dict(calls[0].input) if calls else None


def route(create: Callable[..., Message], raw: str, trades: dict[str, dict[str, str]], model: str) -> Routed:
    email = message_from_string(raw)
    subject = str(email["Subject"])
    result = classify(create, email, model)
    if result is None or result.get("category") not in CATEGORIES:
        return Routed(subject, "unclassified", "ops-triage", {"reason": "model did not call the tool"})
    category = str(result["category"])
    if category == "ssi_change":  # fixed rule, never the model's choice
        return Routed(subject, category, "ssi-verification-callback",
                      {"rule": "verify by call-back to a known number before any change"})
    context: dict[str, str] = {}
    trade_ref = result.get("trade_ref")
    if isinstance(trade_ref, str):
        context = trades.get(trade_ref, {"lookup": f"{trade_ref} not found in reference data"})
    return Routed(subject, category, QUEUES[category], context)


class StubMessages:
    """Returns real anthropic.types.Message objects. Scripted by subject line for a repeatable test."""

    SCRIPT = {
        "Allocation query": {"category": "trade_query", "trade_ref": "TRD-88302", "account_ref": None},
        "Updated bank details": {"category": "ssi_change", "trade_ref": None, "account_ref": "ACC-11893"},
        "Valuation": {"category": "valuation_request", "trade_ref": None, "account_ref": "ACC-20417"},
    }

    def create(self, **kwargs: object) -> Message:
        messages = kwargs["messages"]
        assert isinstance(messages, list)
        text = str(messages[0]["content"])
        hit = next(((k, v) for k, v in self.SCRIPT.items() if k in text), None)
        content: list[dict[str, object]] = (
            [{"type": "tool_use", "id": "toolu_01", "name": "classify_email", "input": hit[1]}] if hit
            else [{"type": "text", "text": "I am not sure how to classify this email."}])
        return Message.model_validate({
            "id": "msg_stub", "type": "message", "role": "assistant", "model": "claude-haiku-5-5",
            "content": content, "stop_reason": "tool_use" if hit else "end_turn", "stop_sequence": None,
            "usage": {"input_tokens": 300, "output_tokens": 40}})


INBOX = [
    "Subject: Allocation query TRD-88302\n\nWe show 1,000 not 1,500 on the second fill, please confirm.",
    "Subject: Updated bank details\n\nPlease use our new account for settlement from Monday.",
    "Subject: Valuation for year end\n\nCould you send the 30 September valuation for ACC-20417?",
    "Subject: Lunch on Friday?\n\nAre you free?",
]
TRADES = {"TRD-88302": {"trade_ref": "TRD-88302", "counterparty": "Sample Broker LLP", "status": "unconfirmed"}}

if __name__ == "__main__":
    stub = StubMessages()
    for raw in INBOX:
        r = route(stub.create, raw, TRADES, model="claude-haiku-5-5")
        print(f"{r.subject[:28]:<28} {r.category:<18} -> {r.queue:<26} {r.context}")

Requires anthropic 1.14.0. Tested on Python 3.14.4 and 3.11.14 with mypy 2.4.0 --strict. Stub client, sample emails.

"""Measure triage accuracy on labelled emails before anyone relies on it.

Computes precision and recall per category from a parallel run (the agent's label against the label the
team gave the same email), and decides which categories may be routed automatically. A category stays
with a person until it clears the precision bar on enough emails. ssi_change never auto-routes.

Standard library only. Tested on Python 3.14.4 and 3.11.14 with mypy --strict. Sample labels only.
"""

from __future__ import annotations

import csv
import io
from collections import Counter
from dataclasses import dataclass

MIN_PRECISION = 0.98
MIN_SUPPORT = 50  # labelled emails per category before auto-routing is even considered
NEVER_AUTO = {"ssi_change"}


@dataclass(frozen=True)
class CategoryResult:
    category: str
    support: int
    precision: float
    recall: float
    auto_route: bool


def evaluate(rows: list[tuple[str, str]]) -> list[CategoryResult]:
    """rows: (team_label, agent_label) pairs from the parallel run."""
    tp: Counter[str] = Counter()
    predicted: Counter[str] = Counter()
    actual: Counter[str] = Counter()
    for team, agent in rows:
        actual[team] += 1
        predicted[agent] += 1
        if team == agent:
            tp[team] += 1
    results = []
    for cat in sorted(actual):
        precision = tp[cat] / predicted[cat] if predicted[cat] else 0.0
        recall = tp[cat] / actual[cat]
        auto = (cat not in NEVER_AUTO and actual[cat] >= MIN_SUPPORT and precision >= MIN_PRECISION)
        results.append(CategoryResult(cat, actual[cat], round(precision, 3), round(recall, 3), auto))
    return results


SAMPLE_CSV = (
    "team_label,agent_label\n"
    + "trade_query,trade_query\n" * 118 + "trade_query,other\n" * 2
    + "document_request,document_request\n" * 61 + "document_request,valuation_request\n" * 3
    + "valuation_request,valuation_request\n" * 40
    + "ssi_change,ssi_change\n" * 12
    + "other,other\n" * 30 + "other,trade_query\n" * 1
)

if __name__ == "__main__":
    reader = csv.DictReader(io.StringIO(SAMPLE_CSV))
    rows = [(r["team_label"], r["agent_label"]) for r in reader]
    print(f"{'category':<18} {'n':>4} {'precision':>9} {'recall':>7}  route")
    for r in evaluate(rows):
        print(f"{r.category:<18} {r.support:>4} {r.precision:>9.3f} {r.recall:>7.3f}  "
              f"{'automatic' if r.auto_route else 'person'}")

Standard library only. Tested on Python 3.14.4 and 3.11.14 with mypy 2.4.0 --strict. Sample labels.