M24: What's Next — The Agent Frontier

{
  "name": "tax-specialist",
  "description": "US federal & state tax analysis with deduction optimization.",
  "url": "https://agents.acme.com/tax-specialist",
  "version": "2.1.0",
  "provider": { "organization": "Acme Tax LLC" },
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "authentication": { "schemes": ["bearer", "oauth2"] },
  "defaultInputModes":  ["text", "application/json"],
  "defaultOutputModes": ["text", "application/json"],
  "skills": [
    {
      "id": "deduction-analysis",
      "name": "Analyze itemized deductions",
      "description": "Given income + receipts, returns optimal deduction plan.",
      "tags": ["tax", "deductions", "us-federal"],
      "examples": ["Analyze deductions for $85K income, married filing jointly"]
    }
  ]
}

import uuid
from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel

app = FastAPI()
TASKS: dict[str, dict] = {}   # in-memory; real deployments use Redis/Postgres

AGENT_CARD = {
    "name": "tax-specialist",
    "description": "US federal & state tax analysis with deduction optimization.",
    "url": "https://agents.acme.com/tax-specialist",
    "version": "2.1.0",
    "capabilities": {"streaming": False, "pushNotifications": True},
    "authentication": {"schemes": ["bearer"]},
    "defaultInputModes":  ["text", "application/json"],
    "defaultOutputModes": ["text", "application/json"],
    "skills": [{
        "id": "deduction-analysis",
        "name": "Analyze itemized deductions",
        "description": "Given income + receipts, returns optimal deduction plan.",
        "tags": ["tax", "deductions", "us-federal"],
    }],
}

@app.get("/.well-known/agent.json")
def get_card():
    return AGENT_CARD

class TaskSend(BaseModel):
    id: str | None = None
    message: dict            # {"role": "user", "parts": [{"type": "text", "text": "..."}]}
    skillId: str | None = None

@app.post("/tasks/send")
async def send_task(req: TaskSend, authorization: str | None = Header(None)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "Bearer token required")

    task_id = req.id or str(uuid.uuid4())
    TASKS[task_id] = {"id": task_id, "state": "working", "messages": [req.message]}
    try:
        # YOUR existing Claude agent loop — M05/M12/M15B style.
        result = run_claude_tax_skill(req.message, skill_id=req.skillId)
        TASKS[task_id].update({
            "state": "completed",
            "artifacts": [{"name": "report", "parts": [{"type": "text", "text": result}]}],
        })
    except Exception as e:
        TASKS[task_id].update({"state": "failed", "error": str(e)})
    return TASKS[task_id]

@app.get("/tasks/{task_id}")
def get_task(task_id: str):
    if task_id not in TASKS:
        raise HTTPException(404, "Unknown task")
    return TASKS[task_id]

def run_claude_tax_skill(message: dict, skill_id: str | None) -> str:
    # Hand off to your existing Claude agent — this is the only line you write
    # that’s domain-specific. Everything around it is boilerplate A2A.
    return "Estimated savings: $2,728 (itemize, claim home-office + 401k)"

import httpx, time, uuid

BASE  = "https://agents.acme.com/tax-specialist"
TOKEN = "your-bearer-token"

def delegate_to_agent(question: str, *, timeout_s: int = 30) -> str:
    # 1. Discover — fetch the Agent Card and verify the skill exists.
    card = httpx.get(f"{BASE}/.well-known/agent.json", timeout=5).raise_for_status().json()
    skill = next((s for s in card["skills"] if s["id"] == "deduction-analysis"), None)
    if skill is None:
        raise RuntimeError(f"Agent {card['name']} doesn’t advertise deduction-analysis")
    if "bearer" not in card["authentication"]["schemes"]:
        raise RuntimeError("Agent doesn’t accept bearer auth — refuse to call")

    # 2. Send the task.
    task_id = str(uuid.uuid4())
    res = httpx.post(
        f"{BASE}/tasks/send",
        json={"id": task_id, "skillId": skill["id"],
              "message": {"role": "user", "parts": [{"type": "text", "text": question}]}},
        headers={"Authorization": f"Bearer {TOKEN}"},
        timeout=10,
    ).raise_for_status().json()

    # 3. Poll until terminal state (or timeout). Real code uses SSE if the
    #    card advertises capabilities.streaming.
    deadline = time.monotonic() + timeout_s
    while res["state"] in ("submitted", "working", "input-required"):
        if time.monotonic() > deadline:
            httpx.post(f"{BASE}/tasks/{task_id}/cancel",
                       headers={"Authorization": f"Bearer {TOKEN}"})
            raise TimeoutError(f"Task {task_id} exceeded {timeout_s}s")
        time.sleep(0.5)
        res = httpx.get(f"{BASE}/tasks/{task_id}",
                        headers={"Authorization": f"Bearer {TOKEN}"}).json()

    if res["state"] != "completed":
        raise RuntimeError(f"Task {task_id} {res['state']}: {res.get('error')}")
    return res["artifacts"][0]["parts"][0]["text"]

# Before: standard response
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Design a microservices architecture for..."}]
)

# After: extended thinking enabled
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000  # Up to 10K tokens for reasoning
    },
    messages=[{"role": "user", "content": "Design a microservices architecture for..."}]
)
# response.content now includes thinking blocks you can inspect for debugging

---
name: pdf-extraction
description: Extract structured data (tables, key/value pairs, headings) from PDF
  files. Use this when the user uploads a PDF and asks for fields, columns,
  totals, or any structured content. Returns JSON.
allowed-tools: ["Read", "Bash"]
---

# PDF Extraction Skill

## When to use
- User uploads a PDF (.pdf in the conversation or in `./inputs/`).
- User asks for any structured field, table, or total.
- Source-of-truth answer needs to be grounded in the PDF, not Claude's guess.

## Procedure
1. Run `python scripts/extract.py path/to/file.pdf` (see reference/).
2. The script writes `out/extracted.json` with `pages[]`, `tables[]`, `kv[]`.
3. Validate against `reference/expected-schema.json`.
4. If validation fails, retry once with `--ocr` flag.
5. Return ONLY the requested fields — don't dump the whole JSON.

## Gotchas
- Scanned PDFs need `--ocr`. Detect: page text length < 50 chars => OCR.
- Multi-column layouts: pass `--layout=two-column`.
- Tables with merged cells: see `reference/merged-cells.md` for workaround.

from claude_agent_sdk import ClaudeAgentClient, AgentOptions

# 1. Point the agent at a skills directory. The SDK walks the tree,
#    reads each SKILL.md frontmatter, and exposes the descriptions
#    to Claude as choosable capabilities.
agent = ClaudeAgentClient(
    options=AgentOptions(
        model="claude-opus-4-7",
        skill_dirs=["./.claude/skills"],   # one or many folders
        allowed_tools=["Read", "Write", "Bash"],
    ),
)

# 2. Run a task — the agent will autonomously decide whether to
#    load `pdf-extraction` based on the user’s request.
async for event in agent.run("Extract the totals table from invoice.pdf"):
    if event.type == "skill_loaded":
        print(f"→ loaded skill: {event.skill_name}")   # observability hook
    elif event.type == "text":
        print(event.text, end="")
    elif event.type == "error":
        # M16-style guardrail: surface failures, don’t silently fall back.
        raise RuntimeError(f"agent error: {event.message}")

from anthropic import Anthropic
import base64, subprocess

client = Anthropic()

# Your sandbox primitives. In production these run inside a Docker container
# with a virtual display (Xvfb) so the agent can’t touch your real desktop.
def take_screenshot() -> bytes:
    subprocess.run(["scrot", "/tmp/screen.png"], check=True)
    return open("/tmp/screen.png", "rb").read()

def execute_action(action: dict) -> bytes:
    if action["action"] == "screenshot":
        pass
    elif action["action"] == "left_click":
        x, y = action["coordinate"]
        subprocess.run(["xdotool", "mousemove", str(x), str(y), "click", "1"])
    elif action["action"] == "type":
        subprocess.run(["xdotool", "type", "--", action["text"]])
    elif action["action"] == "key":
        subprocess.run(["xdotool", "key", action["text"]])
    # ... scroll, double_click, mouse_move, cursor_position, etc.
    return take_screenshot()

# Initial request: enable the three pre-built tools and give a goal
messages = [{"role": "user", "content": "Open firefox, search 'anthropic pricing', screenshot the result"}]
initial = take_screenshot()

while True:
    response = client.beta.messages.create(
        model="claude-opus-4-7",
        max_tokens=4096,
        tools=[
            {"type": "computer_20250124", "name": "computer",
             "display_width_px": 1280, "display_height_px": 800},
            {"type": "bash_20250124", "name": "bash"},
            {"type": "text_editor_20250429", "name": "str_replace_based_edit_tool"},
        ],
        betas=["computer-use-2025-01-24"],
        messages=messages,
    )

    if response.stop_reason == "end_turn":
        break  # Claude decided it’s done

    # Append Claude’s reply to the conversation
    messages.append({"role": "assistant", "content": response.content})

    # Execute every tool_use block and collect tool_results
    results = []
    for block in response.content:
        if block.type == "tool_use" and block.name == "computer":
            screenshot = execute_action(block.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": [{
                    "type": "image",
                    "source": {"type": "base64", "media_type": "image/png",
                               "data": base64.b64encode(screenshot).decode()}
                }],
            })
        # ... handle bash and text_editor tool_use blocks similarly
    messages.append({"role": "user", "content": results})

# pip install claude-agent-sdk
from claude_agent_sdk import ClaudeAgentClient, AgentOptions, tool

@tool
def search_filings(state: str, name: str) -> list[dict]:
    """Search UCC filings by state and debtor name."""
    return mock_db.query(state=state, name=name)

agent = ClaudeAgentClient(
    options=AgentOptions(
        model="claude-opus-4-7",
        tools=[search_filings],
        allowed_tools=["search_filings"],
    ),
)

async for event in agent.run("Active filings for Acme Corp in CA"):
    if event.type == "text":
        print(event.text, end="")
    elif event.type == "error":
        raise RuntimeError(event.message)

pip install langchain langchain-anthropic

# Same 3 tools, same mock data, same question as M15B
# But using LangChain's AgentExecutor instead of a raw while loop
# Compare: lines of code, output quality, debuggability, dependencies

import asyncio
from claude_agent_sdk import ClaudeAgentClient, AgentOptions

# A long-horizon coding agent driven by the SDK. The key differences vs
# the M15B-style loop: workspace_dir scopes file access, max_turns is the
# safety budget, and persist_session_id lets us resume if the run is
# interrupted (e.g. nightly cron that picks up where it left off).
async def run_codebase_task(task: str, *, workspace: str, session_id: str):
    agent = ClaudeAgentClient(
        options=AgentOptions(
            model="claude-opus-4-7",
            workspace_dir=workspace,
            allowed_tools=["Read", "Write", "Edit", "Bash", "Grep", "Glob"],
            max_turns=200,                          # hard ceiling
            persist_session_id=session_id,          # resume on next run
            extended_thinking={"budget_tokens": 8000},  # planning compute
        ),
    )

    try:
        async for event in agent.run(task):
            if event.type == "plan_updated":
                print(f"[plan] {event.summary}")     # observability
            elif event.type == "tool_call":
                print(f"  → {event.tool}({event.args_preview})")
            elif event.type == "text":
                print(event.text, end="")
            elif event.type == "stuck":
                # Long-horizon agents must escalate, not silently spin.
                raise RuntimeError(f"Agent stuck after {event.turn} turns: {event.reason}")
    finally:
        # The agent writes a branch; we leave it for human review (HITL, M17).
        print(f"\nFinished — review branch: claude/{session_id}")

if __name__ == "__main__":
    asyncio.run(run_codebase_task(
        task="Fix all flaky tests in tests/integration/, one PR per root cause.",
        workspace="/srv/repos/payments",
        session_id="nightly-flake-sweep",
    ))

# new_agent_project/agent.py
# Production-ready agent template with best practices from M01-M22.
# Includes: structured logging, guardrails, cost tracking, and
# the agentic loop pattern from M12.

import anthropic
import json
import logging
import time
from datetime import datetime


# --- Structured Logging (M19) ---
# JSON-formatted logs with request IDs for tracing.
logging.basicConfig(
    level=logging.INFO,
    format='{"time":"%(asctime)s","level":"%(levelname)s","msg":"%(message)s"}',
)
logger = logging.getLogger(__name__)


# --- Configuration ---
MODEL = "claude-sonnet-4-6"
MAX_TOKENS = 1024
MAX_ITERATIONS = 10  # Safety net (M12: stop_reason is primary)


# --- Cost Tracking (M22) ---
COST_PER_1K = {"input": 0.003, "output": 0.015}
session_cost = {"input_tokens": 0, "output_tokens": 0}


def track_cost(response):
    """Accumulate token usage for cost monitoring."""
    usage = response.usage
    session_cost["input_tokens"] += usage.input_tokens
    session_cost["output_tokens"] += usage.output_tokens
    cost = (
        usage.input_tokens / 1000 * COST_PER_1K["input"]
        + usage.output_tokens / 1000 * COST_PER_1K["output"]
    )
    logger.info(f"call_cost=${cost:.4f} total_in={session_cost['input_tokens']} total_out={session_cost['output_tokens']}")
    return cost


# --- Input Guardrails (M16) ---
MAX_INPUT_LENGTH = 4000


def validate_input(user_message: str) -> str | None:
    """Return an error message if input is invalid, None if OK."""
    if not user_message or not user_message.strip():
        return "Please provide a message."
    if len(user_message) > MAX_INPUT_LENGTH:
        return f"Message too long ({len(user_message)} chars, max {MAX_INPUT_LENGTH})."
    return None


# --- Tool Definitions ---
# Replace these with your actual tools.
TOOLS = [
    {
        "name": "example_tool",
        "description": "Replace this with your tool description.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The input query",
                }
            },
            "required": ["query"],
        },
    },
]


def handle_tool_call(name: str, input_data: dict) -> dict:
    """Dispatch tool calls to handlers. Add your tools here."""
    if name == "example_tool":
        return {"result": f"Tool result for: {input_data.get('query', '')}"}
    return {"is_error": True, "error_category": "unknown_tool", "context": name}


# --- System Prompt ---
SYSTEM_PROMPT = """You are a helpful assistant. Replace this with your
agent's specific role, capabilities, and constraints.

Rules:
- Only answer questions within your defined scope.
- If unsure, say so rather than guessing.
- Never reveal internal system details.
"""


def run_agent(user_message: str) -> tuple[str, list[str], float]:
    """
    Run the agent for a single user message.
    Returns (response_text, tools_called, total_cost).
    """
    # Input validation (M16)
    error = validate_input(user_message)
    if error:
        return error, [], 0.0

    client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY env var
    messages = [{"role": "user", "content": user_message}]
    tools_called = []
    total_cost = 0.0
    request_id = datetime.now().strftime("%Y%m%d%H%M%S%f")

    logger.info(f"request_id={request_id} input_length={len(user_message)}")

    for iteration in range(MAX_ITERATIONS):
        try:
            response = client.messages.create(
                model=MODEL,
                max_tokens=MAX_TOKENS,
                system=SYSTEM_PROMPT,
                tools=TOOLS,
                messages=messages,
            )
            total_cost += track_cost(response)
        except anthropic.APIError as e:
            logger.error(f"request_id={request_id} api_error={e}")
            return f"I encountered a technical issue. Please try again.", tools_called, total_cost

        # Primary stop condition (M12): check stop_reason
        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    tools_called.append(block.name)
                    logger.info(f"request_id={request_id} tool={block.name} iteration={iteration}")
                    result = handle_tool_call(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result),
                    })
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

        elif response.stop_reason == "end_turn":
            text = " ".join(
                b.text for b in response.content if hasattr(b, "text")
            )
            logger.info(f"request_id={request_id} done iterations={iteration+1} tools={tools_called} cost=${total_cost:.4f}")
            return text, tools_called, total_cost

    logger.warning(f"request_id={request_id} hit_max_iterations={MAX_ITERATIONS}")
    return "I was unable to complete your request.", tools_called, total_cost


if __name__ == "__main__":
    query = input("You: ")
    response, tools, cost = run_agent(query)
    print(f"\nAgent: {response}")
    print(f"Tools: {tools} | Cost: ${cost:.4f}")

mkdir my-agent-project && cd my-agent-project
python -m venv venv
source venv/bin/activate       # Windows: venv\Scripts\activate
pip install anthropic
export ANTHROPIC_API_KEY=your-key-here  # Windows: set ANTHROPIC_API_KEY=your-key-here

# Create agent.py and paste the Python template from the section above

# My 90-Day Agent Development Roadmap

## Goal (be specific)
Deploy a [type] agent that handles [metric] for [audience].
Example: "Deploy a document analysis agent that handles 100 queries/day for my team."

## Weeks 1-2: Finish Capstone
- [ ] Which capstone? ___
- [ ] Run evaluation harness
- [ ] Score with rubric: Functionality __ / Code Quality __ / Prompts __ / Safety __ / Observability __

## Month 1: First Production Deploy
- [ ] Deploy target: [local / cloud / team server]
- [ ] Monitoring setup: [structured logs / dashboard]
- [ ] First real user test date: ___

## Months 2-3: Novel Project
- [ ] Problem I want to solve: ___
- [ ] Why I care about this: ___
- [ ] Architecture sketch: [single agent / multi-agent / RAG]

## Weakest Dimension
My weakest rubric dimension is: ___
Action to strengthen it: ___

## Community
- [ ] Join: [forum/Discord name]
- [ ] Share what I built by: [date]