CC15: Building Custom Agents with the Claude Agent SDK

"""Run any agent with full message-stream visibility."""
import asyncio
import json
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    SystemMessage, AssistantMessage, UserMessage, ResultMessage,
    TextBlock, ToolUseBlock, ToolResultBlock, ThinkingBlock,
)


async def run_agent(prompt: str, options: ClaudeAgentOptions) -> None:
    print(f"\n>>> PROMPT: {prompt}\n")

    async for msg in query(prompt=prompt, options=options):

        if isinstance(msg, SystemMessage):
            print(f"[system] session={msg.data.get('session_id')[:8]}... "
                  f"model={msg.data.get('model')} "
                  f"tools={len(msg.data.get('tools', []))}")

        elif isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(f"[claude] {block.text}")
                elif isinstance(block, ToolUseBlock):
                    args = json.dumps(block.input, indent=None)[:120]
                    print(f"[tool-call] {block.name}({args})")
                elif isinstance(block, ThinkingBlock):
                    print(f"[thinking] {block.thinking[:200]}...")

        elif isinstance(msg, UserMessage):
            for block in msg.content:
                if isinstance(block, ToolResultBlock):
                    body = str(block.content)[:200]
                    status = "ERROR" if block.is_error else "ok"
                    print(f"[tool-result {status}] {body}")

        elif isinstance(msg, ResultMessage):
            print(f"\n[done] turns={msg.num_turns} "
                  f"duration={msg.duration_ms}ms "
                  f"cost=${msg.total_cost_usd:.4f} "
                  f"error={msg.is_error}")


if __name__ == "__main__":
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Glob"],
        max_turns=5,
        permission_mode="bypassPermissions",
    )
    asyncio.run(run_agent("Summarize files in this dir.", options))

>>> PROMPT: Summarize files in this dir.

[system] session=a3f2c8d1... model=claude-opus-4-7 tools=14
[tool-call] Glob({"pattern": "*"})
[tool-result ok] README.md
deps.txt
first_agent.py
[tool-call] Read({"file_path": "/Users/me/cc-labs/first-agent/README.md"})
[tool-result ok] # My first agent project
[tool-call] Read({"file_path": "/Users/me/cc-labs/first-agent/deps.txt"})
[tool-result ok] claude-agent-sdk
[claude] This directory contains a starter project using the Claude Agent SDK.
The README labels it "My first agent project" and deps.txt declares one
dependency: claude-agent-sdk.

[done] turns=4 duration=3287ms cost=$0.0093 error=false

import json, time
from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions

trace_file = Path(f"trace-{int(time.time())}.jsonl")

async def traced(prompt, options):
    with trace_file.open("w") as f:
        async for msg in query(prompt=prompt, options=options):
            # SDK message types are dataclasses with .__dict__-friendly contents.
            entry = {"ts": time.time(), "type": type(msg).__name__,
                     "raw": repr(msg)[:2000]}
            f.write(json.dumps(entry) + "\n")
            f.flush()  # crash-safe
    print(f"trace written to {trace_file}")

from claude_agent_sdk import HookMatcher, ClaudeAgentOptions

async def log_tool_call(input_data, tool_use_id, context):
    print(f"[hook] {input_data['tool_name']} -> {input_data['tool_input']}")
    return {}  # empty dict = allow, no modifications

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [HookMatcher(matcher="*", hooks=[log_tool_call])],
    },
)

"""Your first Claude Agent SDK program."""
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock


async def main() -> None:
    options = ClaudeAgentOptions(
        system_prompt="You are a concise project explainer.",
        allowed_tools=["Read", "Glob"],
        max_turns=4,
        permission_mode="bypassPermissions",
    )

    prompt = "Read the files in this directory and summarize the project in 2 sentences."

    async for msg in query(prompt=prompt, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text, end="", flush=True)
    print()


if __name__ == "__main__":
    asyncio.run(main())

// Your first Claude Agent SDK program.
import { query } from "@anthropic-ai/claude-agent-sdk";

const stream = query({
  prompt: "Read the files in this directory and summarize the project in 2 sentences.",
  options: {
    systemPrompt: "You are a concise project explainer.",
    allowedTools: ["Read", "Glob"],
    maxTurns: 4,
    permissionMode: "bypassPermissions",
  },
});

for await (const msg of stream) {
  if (msg.type === "assistant") {
    for (const block of msg.message.content) {
      if (block.type === "text") process.stdout.write(block.text);
    }
  }
}
process.stdout.write("\n");

This project is a starter scaffold for working with the Claude Agent SDK,
described in README.md. Its only declared dependency is `claude-agent-sdk`,
listed in deps.txt.

from claude_agent_sdk import ToolUseBlock

# add this branch alongside the TextBlock branch:
if isinstance(block, ToolUseBlock):
    print(f"\n[tool: {block.name}({block.input})]\n", end="", flush=True)

"""UCC Filings MCP server — wraps the PublicRecords API as agent tools."""
import asyncio
import httpx
from claude_agent_sdk import create_sdk_mcp_server, tool

API = "http://localhost:8080"


@tool(
    "lookup_filings",
    "Search UCC filings by 2-letter US state code. Returns up to 50 matches.",
    {"state": str},
)
async def lookup_filings(args: dict) -> dict:
    state = args["state"].upper()
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(f"{API}/filings", params={"state": state})
        r.raise_for_status()
        rows = r.json()
    summary = [
        f"#{f['id']} | {f['state']} | {f['debtorName']} -> {f['securedParty']}"
        for f in rows[:50]
    ]
    return {"content": [{"type": "text", "text": "\n".join(summary) or "no matches"}]}


@tool(
    "get_filing",
    "Fetch the full record of a single UCC filing by numeric id.",
    {"filing_id": int},
)
async def get_filing(args: dict) -> dict:
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.get(f"{API}/filings/{args['filing_id']}")
        if r.status_code == 404:
            return {"content": [{"type": "text", "text": "filing not found"}]}
        r.raise_for_status()
        f = r.json()
    return {"content": [{"type": "text", "text": str(f)}]}


ucc_server = create_sdk_mcp_server(
    name="ucc",
    version="0.1.0",
    tools=[lookup_filings, get_filing],
)

// UCC Filings MCP server — wraps the PublicRecords API as agent tools.
import { createSdkMcpServer, tool } from "@anthropic-ai/claude-agent-sdk";
import { z } from "zod";

const API = "http://localhost:8080";

export const uccServer = createSdkMcpServer({
  name: "ucc",
  version: "0.1.0",
  tools: [
    tool(
      "lookup_filings",
      "Search UCC filings by 2-letter US state code. Returns up to 50 matches.",
      { state: z.string().length(2) },
      async ({ state }) => {
        const r = await fetch(`${API}/filings?state=${state.toUpperCase()}`);
        if (!r.ok) throw new Error(`API ${r.status}`);
        const rows = (await r.json()) as Array<Record<string, unknown>>;
        const summary = rows
          .slice(0, 50)
          .map((f) => `#${f.id} | ${f.state} | ${f.debtorName} -> ${f.securedParty}`)
          .join("\n");
        return { content: [{ type: "text", text: summary || "no matches" }] };
      },
    ),
    tool(
      "get_filing",
      "Fetch the full record of a single UCC filing by numeric id.",
      { filing_id: z.number().int().positive() },
      async ({ filing_id }) => {
        const r = await fetch(`${API}/filings/${filing_id}`);
        if (r.status === 404) return { content: [{ type: "text", text: "filing not found" }] };
        if (!r.ok) throw new Error(`API ${r.status}`);
        const f = await r.json();
        return { content: [{ type: "text", text: JSON.stringify(f, null, 2) }] };
      },
    ),
  ],
});

"""UCC Filings Assistant — agent driver that uses the ucc MCP server."""
import asyncio
import sys
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock
from ucc_mcp import ucc_server

SYSTEM = """You are a UCC Filings Assistant for US public records.

Definitions:
- UCC-1 filing: a public lien notice; a "secured party" (lender) holding
  a security interest in collateral belonging to a "debtor" (borrower).

You have two tools:
- mcp__ucc__lookup_filings(state) — list matches for a US state
- mcp__ucc__get_filing(filing_id) — full record for one filing

Be concise. Always cite filing ids in your answer. If the user asks for a
state by full name, infer the 2-letter code (Texas -> TX).
"""


async def main(prompt: str) -> None:
    options = ClaudeAgentOptions(
        system_prompt=SYSTEM,
        mcp_servers={"ucc": ucc_server},
        allowed_tools=["mcp__ucc__lookup_filings", "mcp__ucc__get_filing"],
        max_turns=6,
        permission_mode="bypassPermissions",
    )

    async for msg in query(prompt=prompt, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text, end="", flush=True)
    print()


if __name__ == "__main__":
    user_q = sys.argv[1] if len(sys.argv) > 1 else "Find me Texas filings."
    asyncio.run(main(user_q))

// UCC Filings Assistant — agent driver that uses the ucc MCP server.
import { query } from "@anthropic-ai/claude-agent-sdk";
import { uccServer } from "./ucc_mcp";

const SYSTEM = `You are a UCC Filings Assistant for US public records.

Definitions:
- UCC-1 filing: a public lien notice; a "secured party" (lender) holding
  a security interest in collateral belonging to a "debtor" (borrower).

You have two tools:
- mcp__ucc__lookup_filings(state) — list matches for a US state
- mcp__ucc__get_filing(filing_id) — full record for one filing

Be concise. Always cite filing ids in your answer. If the user asks for a
state by full name, infer the 2-letter code (Texas -> TX).`;

const prompt = process.argv[2] ?? "Find me Texas filings.";

const stream = query({
  prompt,
  options: {
    systemPrompt: SYSTEM,
    mcpServers: { ucc: uccServer },
    allowedTools: ["mcp__ucc__lookup_filings", "mcp__ucc__get_filing"],
    maxTurns: 6,
    permissionMode: "bypassPermissions",
  },
});

for await (const msg of stream) {
  if (msg.type === "assistant") {
    for (const block of msg.message.content) {
      if (block.type === "text") process.stdout.write(block.text);
    }
  }
}
process.stdout.write("\n");

Based on the Texas filings:

#1 — Lone Star Holdings LLC -> First National Bank
#2 — Pecos River Logistics Inc. -> Wells Fargo Equipment Finance

Each Texas filing has a different secured party — no single lender appears
more than once in the Texas subset. The two distinct secured parties are
First National Bank and Wells Fargo Equipment Finance.

---
name: ucc-assistant
description: Answers natural-language questions about UCC public records
  by querying the local PublicRecords API. Use when the user asks about
  filings, debtors, secured parties, or collateral.
tools: mcp__ucc__lookup_filings, mcp__ucc__get_filing
model: haiku
---

You are a UCC Filings Assistant. You have two tools:
- `mcp__ucc__lookup_filings(state)` — list filings for a US state code
- `mcp__ucc__get_filing(filing_id)` — full record for one filing

Be concise. Always cite filing ids in your answer.

"""Capture an Agent SDK message stream to structured JSONL for the viewer."""
import json, time
from pathlib import Path
from dataclasses import asdict, is_dataclass

TRACE_DIR = Path("traces")
TRACE_DIR.mkdir(exist_ok=True)


def _serialize(obj):
    """Recursively convert SDK message objects to JSON-safe dicts."""
    if is_dataclass(obj):
        return _serialize(asdict(obj))
    if hasattr(obj, "__dict__"):
        return {k: _serialize(v) for k, v in vars(obj).items()
                if not k.startswith("_")}
    if isinstance(obj, list):
        return [_serialize(x) for x in obj]
    if isinstance(obj, dict):
        return {k: _serialize(v) for k, v in obj.items()}
    return obj  # primitives


class TraceRecorder:
    def __init__(self, label: str = "run"):
        self.path = TRACE_DIR / f"{int(time.time())}-{label}.jsonl"
        self._f = self.path.open("w", encoding="utf-8")

    def record(self, msg):
        entry = {
            "ts": time.time(),
            "type": type(msg).__name__,
            "data": _serialize(msg),
        }
        self._f.write(json.dumps(entry, default=str) + "\n")
        self._f.flush()  # so the viewer can tail live runs

    def close(self):
        self._f.close()
        return self.path

from trace_recorder import TraceRecorder

async def main():
    recorder = TraceRecorder(label="ucc-tx-query")
    try:
        async for msg in query(
            prompt="List the first 3 UCC filings in Texas with their debtors.",
            options=options,
        ):
            recorder.record(msg)   # <-- the only new line
            # ... your existing print / stream logic stays the same ...
    finally:
        path = recorder.close()
        print(f"\nTrace written: {path}")

"""Render captured agent traces as an HTML timeline."""
import json
from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse

app = FastAPI()
TRACE_DIR = Path("traces")


@app.get("/api/traces")
def list_traces():
    files = sorted(TRACE_DIR.glob("*.jsonl"), reverse=True)
    return [{"name": f.name, "size": f.stat().st_size} for f in files]


@app.get("/api/trace/{name}")
def read_trace(name: str):
    if ".." in name or "/" in name:
        raise HTTPException(400, "invalid name")
    p = TRACE_DIR / name
    if not p.exists():
        raise HTTPException(404)
    entries = [json.loads(line) for line in p.read_text().splitlines() if line.strip()]
    return JSONResponse(entries)


@app.get("/", response_class=HTMLResponse)
def index():
    return Path(__file__).parent.joinpath("viewer.html").read_text()

<!DOCTYPE html>
<html><head><meta charset="utf-8"><title>Agent Trace Viewer</title>
<style>
  body{font:14px/1.5 system-ui;background:#0f172a;color:#e2e8f0;margin:0;display:flex;}
  aside{width:260px;background:#1e293b;height:100vh;overflow-y:auto;padding:1rem;}
  aside a{display:block;padding:0.4rem;color:#94a3b8;text-decoration:none;border-radius:4px;font-size:0.82rem;}
  aside a:hover, aside a.active{background:#334155;color:#fff;}
  main{flex:1;padding:1.5rem 2rem;overflow-y:auto;height:100vh;}
  .msg{border-left:3px solid #475569;padding:0.6rem 1rem;margin:0.5rem 0;background:#1e293b;border-radius:0 6px 6px 0;}
  .msg.SystemMessage{border-color:#64748b;}
  .msg.AssistantMessage{border-color:#60a5fa;}
  .msg.UserMessage{border-color:#a78bfa;}
  .msg.ResultMessage{border-color:#34d399;background:#022c22;}
  .type{font-size:0.7rem;text-transform:uppercase;letter-spacing:0.08em;color:#94a3b8;font-weight:600;}
  .ts{float:right;color:#64748b;font-size:0.75rem;}
  pre{background:#0f172a;padding:0.6rem;border-radius:4px;overflow-x:auto;font-size:0.78rem;margin:0.4rem 0 0;color:#cbd5e1;}
  .tool{color:#fbbf24;font-weight:600;}
  .cost{color:#34d399;font-weight:600;}
  .banner{padding:0.8rem 1rem;border-radius:6px;margin-bottom:1rem;}
</style></head><body>
<aside id="list"><h3 style="margin-top:0;font-size:0.85rem;color:#94a3b8;">Runs</h3></aside>
<main id="trace"><p style="color:#64748b">Pick a run on the left.</p></main>
<script>
const esc = s => String(s).replace(/[&<>]/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;'}[c]));
async function loadList(){
  const r = await fetch('/api/traces'); const files = await r.json();
  document.getElementById('list').innerHTML +=
    files.map(f => `<a href="#" data-name="${esc(f.name)}">${esc(f.name)}</a>`).join('');
  document.querySelectorAll('aside a').forEach(a => a.onclick = ev => {
    ev.preventDefault();
    document.querySelectorAll('aside a').forEach(x => x.classList.remove('active'));
    a.classList.add('active');
    render(a.dataset.name);
  });
}
function summarize(e){
  const d = e.data;
  if(e.type === 'AssistantMessage' && d.content){
    return d.content.map(c => {
      if(c.type === 'text') return `<div>${esc(c.text || '')}</div>`;
      if(c.type === 'tool_use')
        return `<div class="tool">🔧 ${esc(c.name)}</div><pre>${esc(JSON.stringify(c.input, null, 2))}</pre>`;
      if(c.type === 'thinking')
        return `<em style="color:#94a3b8">thinking: ${esc((c.thinking || '').slice(0,200))}...</em>`;
      return `<pre>${esc(JSON.stringify(c, null, 2))}</pre>`;
    }).join('');
  }
  if(e.type === 'UserMessage' && d.content){
    return d.content.map(c => c.type === 'tool_result'
      ? `<div class="tool">↩ tool_result</div><pre>${esc((typeof c.content === 'string' ? c.content : JSON.stringify(c.content, null, 2)).slice(0,400))}</pre>`
      : `<pre>${esc(JSON.stringify(c, null, 2))}</pre>`
    ).join('');
  }
  if(e.type === 'ResultMessage'){
    return `<div>turns: ${d.num_turns || '?'} · duration: ${((d.duration_ms || 0)/1000).toFixed(2)}s · <span class="cost">cost: $${(d.total_cost_usd || 0).toFixed(4)}</span></div>`;
  }
  return `<pre>${esc(JSON.stringify(d, null, 2).slice(0,300))}</pre>`;
}
async function render(name){
  const r = await fetch('/api/trace/' + encodeURIComponent(name));
  const entries = await r.json();
  const result = entries.find(e => e.type === 'ResultMessage');
  const banner = result
    ? `<div class="banner" style="background:#022c22">✅ ${result.data.num_turns || '?'} turns · ${((result.data.duration_ms || 0)/1000).toFixed(2)}s · <span class="cost">$${(result.data.total_cost_usd || 0).toFixed(4)}</span></div>`
    : `<div class="banner" style="background:#451a03;color:#fbbf24">⏳ Run in progress...</div>`;
  const body = entries.map(e => {
    const t = new Date(e.ts * 1000).toLocaleTimeString();
    return `<div class="msg ${e.type}"><span class="ts">${t}</span><div class="type">${e.type}</div>${summarize(e)}</div>`;
  }).join('');
  document.getElementById('trace').innerHTML = banner + body;
}
loadList();
// Live tail: re-render the active trace every 1.5s
setInterval(() => { const a = document.querySelector('aside a.active'); if(a) render(a.dataset.name); }, 1500);
</script></body></html>