Deploying
Agents

pip install anthropic   # already installed if you did prior labs
touch prod_agent.py

import os, json, time, random, logging
from datetime import datetime, timezone
import anthropic

# ── Structured logger ────────────────────────────────────────────
# Use JSON lines format so every log entry is machine-parseable.
logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("prod_agent")

def jlog(event: str, **kwargs) -> None:
    """Emit a structured JSON log line."""
    record = {"ts": datetime.now(timezone.utc).isoformat(), "event": event, **kwargs}
    log.info(json.dumps(record))

# ── Client ────────────────────────────────────────────────────────
# max_retries=4: SDK retries 429/529 with built-in exponential backoff.
client = anthropic.Anthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    max_retries=4,
)

# ── Budget constants ──────────────────────────────────────────────
MAX_ITERATIONS = 10      # hard iteration cap per run
TOKEN_BUDGET    = 20_000  # total tokens across all calls in one run
MODEL           = "claude-haiku-4-5-20251001"

TOOLS = [
    {
        "name": "web_search",
        "description": "Search the web for factual information.",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    },
    {
        "name": "calculator",
        "description": "Evaluate a Python arithmetic expression.",
        "input_schema": {
            "type": "object",
            "properties": {"expression": {"type": "string"}},
            "required": ["expression"],
        },
    },
]

SEARCH_DB = {
    "population of earth": "Earth's population is approximately 8.1 billion as of 2024.",
    "speed of light": "The speed of light in a vacuum is 299,792,458 metres per second.",
}

def execute_tool(name: str, tool_input: dict) -> str:
    jlog("tool_call", tool=name, input=tool_input)
    t0 = time.monotonic()
    try:
        if name == "web_search":
            q = tool_input["query"].lower()
            for key, val in SEARCH_DB.items():
                if key in q:
                    result = val
                    break
            else:
                result = f'No results for "{tool_input["query"]}".'
        elif name == "calculator":
            result = str(eval(tool_input["expression"], {"__builtins__": {}}))
        else:
            result = f"Unknown tool: {name}"
    except Exception as e:
        result = f"Tool error: {type(e).__name__}: {e}"

    latency_ms = int((time.monotonic() - t0) * 1000)
    jlog("tool_result", tool=name, result=result[:120], latency_ms=latency_ms)
    return result


def llm_call(messages: list, run_state: dict, iteration: int) -> anthropic.types.Message:
    """Thin wrapper: calls the API and logs usage."""
    t0 = time.monotonic()
    response = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        tools=TOOLS,
        messages=messages,
    )
    latency_ms = int((time.monotonic() - t0) * 1000)

    # Accumulate tokens
    run_state["input_tokens"]  += response.usage.input_tokens
    run_state["output_tokens"] += response.usage.output_tokens
    total = run_state["input_tokens"] + run_state["output_tokens"]

    jlog(
        "llm_call",
        iteration=iteration,
        stop_reason=response.stop_reason,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        total_tokens_so_far=total,
        latency_ms=latency_ms,
    )
    return response

SYSTEM = """You are a research assistant. Use tools to answer the user's question.
When you have a complete answer, respond with a final text message."""


def run_agent(task: str, trace_id: str | None = None) -> dict:
    """Run the agent loop with full production instrumentation."""
    trace_id = trace_id or str(random.randint(100_000, 999_999))
    run_state = {"input_tokens": 0, "output_tokens": 0}

    jlog("run_start", trace_id=trace_id, task=task)
    t_run = time.monotonic()

    messages = [{"role": "user", "content": task}]
    final_answer = None
    stop_reason_outer = "unknown"

    for iteration in range(1, MAX_ITERATIONS + 1):

        # ── Token budget check ───────────────────────────────────
        total_so_far = run_state["input_tokens"] + run_state["output_tokens"]
        if total_so_far >= TOKEN_BUDGET:
            stop_reason_outer = "token_budget_exceeded"
            jlog("budget_exceeded", trace_id=trace_id, total_tokens=total_so_far)
            final_answer = (
                f"[Agent stopped: token budget of {TOKEN_BUDGET:,} exceeded at "
                f"{total_so_far:,} tokens. Partial work completed in {iteration-1} iterations.]"
            )
            break

        response = llm_call(messages, run_state, iteration)
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            stop_reason_outer = "end_turn"
            final_answer = "".join(
                b.text for b in response.content if hasattr(b, "text")
            )
            break

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })
            messages.append({"role": "user", "content": tool_results})

    else:
        # for-loop exhausted — iteration cap hit
        stop_reason_outer = "iteration_cap"
        final_answer = (
            f"[Agent stopped: reached {MAX_ITERATIONS} iteration limit without completing the task.]"
        )

    duration_ms = int((time.monotonic() - t_run) * 1000)
    total_tokens = run_state["input_tokens"] + run_state["output_tokens"]

    jlog(
        "run_end",
        trace_id=trace_id,
        stop_reason=stop_reason_outer,
        total_input_tokens=run_state["input_tokens"],
        total_output_tokens=run_state["output_tokens"],
        total_tokens=total_tokens,
        duration_ms=duration_ms,
    )

    return {
        "trace_id": trace_id,
        "answer": final_answer,
        "total_tokens": total_tokens,
        "stop_reason": stop_reason_outer,
    }

EVAL_CASES = [
    {
        "task": "What is the approximate population of Earth?",
        "must_contain": ["8", "billion"],
        "must_not_contain": ["trillion", "million people"],
    },
    {
        "task": "What is 1234 multiplied by 5678?",
        "must_contain": ["7,006,652", "7006652"],
        "must_not_contain": [],
    },
    {
        "task": "What is the speed of light and how many kilometres does it travel in one second?",
        "must_contain": ["299", "792"],
        "must_not_contain": [],
    },
]


def run_eval() -> None:
    print("\n" + "="*60)
    print("EVALUATION RUN")
    print("="*60)

    passed = 0
    for i, case in enumerate(EVAL_CASES, 1):
        print(f"\n[Case {i}] {case['task']}")
        result = run_agent(case["task"])
        answer = (result["answer"] or "").lower()

        ok = (
            all(kw.lower() in answer for kw in case["must_contain"]) and
            all(kw.lower() not in answer for kw in case["must_not_contain"])
        )

        status = "PASS ✓" if ok else "FAIL ✗"
        print(f"  Status:  {status}")
        print(f"  Tokens:  {result['total_tokens']:,}")
        print(f"  Answer:  {result['answer'][:120]}")
        if ok:
            passed += 1

    print(f"\nResult: {passed}/{len(EVAL_CASES)} cases passed.")
    print("="*60)

if __name__ == "__main__":
    # Single run with a trace ID
    result = run_agent(
        "What is the population of Earth, and how many people is that per square kilometre of land area? Land area = 148,940,000 km².",
        trace_id="demo-001",
    )
    print(f"\nFINAL ANSWER:\n{result['answer']}")
    print(f"Trace: {result['trace_id']} | Tokens: {result['total_tokens']:,} | Stop: {result['stop_reason']}")

    # Eval harness
    run_eval()

python prod_agent.py 2>&1 | tee run.jsonl

{"ts": "2026-04-04T10:00:00Z", "event": "run_start", "trace_id": "demo-001", "task": "What is the population..."}
{"ts": "...", "event": "llm_call", "iteration": 1, "stop_reason": "tool_use", "input_tokens": 412, "output_tokens": 48, "total_tokens_so_far": 460, "latency_ms": 621}
{"ts": "...", "event": "tool_call", "tool": "web_search", "input": {"query": "population of earth"}}
{"ts": "...", "event": "tool_result", "tool": "web_search", "result": "Earth's population is approximately 8.1 billion...", "latency_ms": 0}
{"ts": "...", "event": "llm_call", "iteration": 2, "stop_reason": "tool_use", "input_tokens": 521, "output_tokens": 54, "total_tokens_so_far": 1035, "latency_ms": 588}
{"ts": "...", "event": "tool_call", "tool": "calculator", "input": {"expression": "8100000000 / 148940000"}}
{"ts": "...", "event": "tool_result", "tool": "calculator", "result": "54.38...", "latency_ms": 0}
{"ts": "...", "event": "llm_call", "iteration": 3, "stop_reason": "end_turn", "input_tokens": 630, "output_tokens": 72, "total_tokens_so_far": 1737, "latency_ms": 701}
{"ts": "...", "event": "run_end", "trace_id": "demo-001", "stop_reason": "end_turn", "total_tokens": 1737, "duration_ms": 1950}

import functools

def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 30.0):
    """Decorator: retry a function on exception with exponential backoff + jitter."""
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries:
                        raise
                    delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
                    jlog("tool_retry", attempt=attempt + 1, error=str(e), delay_s=round(delay, 2))
                    time.sleep(delay)
        return wrapper
    return decorator


# Usage: decorate any external tool call that may fail transiently
@retry_with_backoff(max_retries=3)
def call_external_api(endpoint: str) -> str:
    # Replace with a real requests.get() or httpx call
    if random.random() < 0.5:
        raise ConnectionError("Simulated transient failure")
    return f"Response from {endpoint}"