Agent SDKs and the Patterns Underneath

# script_id: day_096_claude_agent_sdk/support_agent
import anthropic

client = anthropic.Anthropic()


# 1. Tool implementations (plain Python functions)
def search_web(query: str) -> str:
    """Pretend web search — wire to a real API in production."""
    return f"Results for '{query}': [result 1], [result 2]"


def calculate(expression: str) -> str:
    """Evaluate arithmetic safely (never eval untrusted input)."""
    import ast, operator
    ops = {ast.Add: operator.add, ast.Sub: operator.sub,
           ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow}

    def _eval(node):
        if isinstance(node, ast.Constant):
            return node.value
        if isinstance(node, ast.BinOp):
            return ops[type(node.op)](_eval(node.left), _eval(node.right))
        if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
            return -_eval(node.operand)
        raise ValueError("Unsupported expression")

    try:
        return str(_eval(ast.parse(expression, mode="eval").body))
    except Exception as e:  # noqa: BLE001
        return f"Error: {e}"


TOOL_IMPLS = {"search_web": search_web, "calculate": calculate}

# 2. Tool schemas the model sees
TOOLS = [
    {
        "name": "search_web",
        "description": "Search the web for current information.",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "Search query"}},
            "required": ["query"],
        },
    },
    {
        "name": "calculate",
        "description": "Evaluate a math expression like '100 * 0.15'.",
        "input_schema": {
            "type": "object",
            "properties": {"expression": {"type": "string"}},
            "required": ["expression"],
        },
    },
]


# 3. The loop the SDKs automate for you
def run_agent(system: str, tools: list, impls: dict, user_message: str,
              model: str = "claude-opus-4-8", max_steps: int = 6) -> str:
    messages = [{"role": "user", "content": user_message}]
    for _ in range(max_steps):
        response = client.messages.create(
            model=model,
            max_tokens=1024,
            system=system,
            tools=tools,
            messages=messages,
        )
        if response.stop_reason != "tool_use":
            # No more tool calls — return the final text.
            return "".join(b.text for b in response.content if b.type == "text")

        # Echo the assistant turn (including its tool_use blocks), then run the tools.
        messages.append({"role": "assistant", "content": response.content})
        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = impls[block.name](**block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(output),
                })
        # Tool results go back under role="user" — that is just how the Messages API
        # carries tool output to the model; it is not literally from the human.
        messages.append({"role": "user", "content": results})
    return "Stopped: max steps reached."


print(run_agent(
    system="You are a research assistant. Use tools for facts and math.",
    tools=TOOLS,
    impls=TOOL_IMPLS,
    user_message="What is 15% of 847?",
))

# script_id: day_096_claude_agent_sdk/support_agent

SPECIALISTS = {
    "billing": {
        "system": "You are a billing specialist. Be precise with amounts and dates; "
                  "confirm before processing any change.",
        "model": "claude-haiku-4-5",
    },
    "technical": {
        "system": "You are technical support. Ask for error messages and stack traces "
                  "when relevant.",
        "model": "claude-sonnet-4-6",
    },
    "general": {
        "system": "You handle general product and account questions.",
        "model": "claude-haiku-4-5",
    },
}


def triage(user_message: str) -> str:
    """Force the model to pick a specialist via a single-tool choice."""
    route_tool = {
        "name": "route",
        "description": "Route the request to the right specialist.",
        "input_schema": {
            "type": "object",
            "properties": {
                "specialist": {"type": "string", "enum": list(SPECIALISTS)},
            },
            "required": ["specialist"],
        },
    }
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=256,
        system="Route the customer's request. Do not answer it yourself.",
        tools=[route_tool],
        tool_choice={"type": "tool", "name": "route"},  # must call route
        messages=[{"role": "user", "content": user_message}],
    )
    tool_use = next(b for b in response.content if b.type == "tool_use")
    return tool_use.input["specialist"]


def handle(user_message: str) -> str:
    specialist = triage(user_message)
    cfg = SPECIALISTS[specialist]
    print(f"Routed to: {specialist}")
    return run_agent(
        system=cfg["system"], tools=TOOLS, impls=TOOL_IMPLS,
        user_message=user_message, model=cfg["model"],
    )


print(handle("I was charged twice for my subscription this month"))

# script_id: day_096_claude_agent_sdk/guardrails
import re


class GuardrailTripped(Exception):
    """Raised when a guardrail blocks the request or response."""


PII_PATTERNS = {
    "SSN": r"\b\d{3}-\d{2}-\d{4}\b",
    "Credit Card": r"\b(?:4[0-9]{12}(?:[0-9]{3})?|5[1-5][0-9]{14})\b",
}


def input_guardrail(message: str) -> None:
    """Block PII before it ever reaches the model."""
    for label, pattern in PII_PATTERNS.items():
        if re.search(pattern, message):
            raise GuardrailTripped(f"Input blocked: contains {label}")


def output_guardrail(response_text: str) -> None:
    """Block low-quality or injection-echoing responses."""
    if len(response_text.strip()) < 20:
        raise GuardrailTripped("Output blocked: too short to be useful")
    for pattern in ("ignore previous instructions", "jailbreak"):
        if pattern in response_text.lower():
            raise GuardrailTripped(f"Output blocked: contains '{pattern}'")


def guarded_run(run_fn, user_message: str) -> str:
    """Wrap any agent call with input/output guardrails."""
    input_guardrail(user_message)            # pre-check
    result = run_fn(user_message)            # run the agent
    output_guardrail(result)                 # post-check
    return result


# Usage:
#   guarded_run(handle, "How do I reset my password?")  -> runs
#   guarded_run(handle, "My SSN is 123-45-6789")        -> GuardrailTripped

# script_id: day_096_claude_agent_sdk/async_api
import anthropic
from fastapi import FastAPI
from pydantic import BaseModel

aclient = anthropic.AsyncAnthropic()
app = FastAPI()


async def run_agent_async(system: str, tools: list, impls: dict, user_message: str,
                          model: str = "claude-opus-4-8", max_steps: int = 6) -> str:
    messages = [{"role": "user", "content": user_message}]
    for _ in range(max_steps):
        response = await aclient.messages.create(
            model=model, max_tokens=1024, system=system, tools=tools, messages=messages,
        )
        if response.stop_reason != "tool_use":
            return "".join(b.text for b in response.content if b.type == "text")
        messages.append({"role": "assistant", "content": response.content})
        results = [
            {"type": "tool_result", "tool_use_id": b.id, "content": str(impls[b.name](**b.input))}
            for b in response.content if b.type == "tool_use"
        ]
        messages.append({"role": "user", "content": results})
    return "Stopped: max steps reached."


class ChatRequest(BaseModel):
    message: str


@app.post("/chat")
async def chat(request: ChatRequest):
    answer = await run_agent_async(
        system="You are a helpful assistant.",
        tools=[], impls={}, user_message=request.message,
    )
    return {"response": answer}

# script_id: day_096_claude_agent_sdk/support_agent

def run_agent_traced(system: str, tools: list, impls: dict, user_message: str,
                     model: str = "claude-opus-4-8", max_steps: int = 6) -> str:
    messages = [{"role": "user", "content": user_message}]
    for step in range(max_steps):
        response = client.messages.create(
            model=model, max_tokens=1024, system=system, tools=tools, messages=messages,
        )
        u = response.usage
        print(f"[step {step}] stop={response.stop_reason} "
              f"in={u.input_tokens} out={u.output_tokens}")
        for b in response.content:
            if b.type == "tool_use":
                print(f"  tool_use: {b.name}({b.input})")
        if response.stop_reason != "tool_use":
            return "".join(b.text for b in response.content if b.type == "text")
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": [
            {"type": "tool_result", "tool_use_id": b.id, "content": str(impls[b.name](**b.input))}
            for b in response.content if b.type == "tool_use"
        ]})
    return "Stopped: max steps reached."