Capstone 7-C — Agent Evolution: UCC Risk Analyzer

mkdir agent-iter1-raw && cd agent-iter1-raw
python -m venv venv && source venv/bin/activate    # Windows: venv\Scripts\activate
pip install "anthropic>=0.40" "scikit-learn>=1.3" "pandas>=2.0" "fastapi>=0.110" "uvicorn>=0.27"

# Train the delinquency_model.pkl
python -c "
from sklearn.ensemble import RandomForestClassifier
import pickle, numpy as np
np.random.seed(42)
X = np.random.rand(200, 6); y = (X.sum(axis=1) > 3).astype(int)
clf = RandomForestClassifier(n_estimators=20, random_state=42).fit(X, y)
pickle.dump(clf, open('delinquency_model.pkl', 'wb'))
print('Model saved')
"

Model saved
12 filings loaded

"""tools.py — UCC tool schemas + executors for the raw API loop."""
import pickle
from mock_data import FILINGS

TOOLS = [
    {
        "name": "search_filings",
        "description": "Search UCC filings by debtor name (partial, case-insensitive). Optional state filter.",
        "input_schema": {
            "type": "object",
            "properties": {
                "debtor_name": {"type": "string"},
                "state": {"type": "string", "description": "2-letter code (optional)"},
            },
            "required": ["debtor_name"],
        },
    },
    {
        "name": "get_filing_details",
        "description": "Get full details for a specific filing by filing number.",
        "input_schema": {
            "type": "object",
            "properties": {"filing_number": {"type": "string"}},
            "required": ["filing_number"],
        },
    },
    {
        "name": "predict_delinquency",
        "description": "Run the ML delinquency model. Returns probability and HIGH/MEDIUM/LOW.",
        "input_schema": {
            "type": "object",
            "properties": {
                "active_filing_count": {"type": "integer"},
                "state_count": {"type": "integer"},
                "collateral_types": {"type": "integer"},
                "filing_age_years": {"type": "number"},
                "amendment_frequency": {"type": "number"},
                "months_to_lapse": {"type": "number"},
            },
            "required": ["active_filing_count", "state_count", "collateral_types",
                         "filing_age_years", "amendment_frequency", "months_to_lapse"],
        },
    },
]

_MODEL = pickle.load(open("delinquency_model.pkl", "rb"))

def execute_tool(name, args):
    if name == "search_filings":
        q = args["debtor_name"].lower()
        st = args.get("state", "").upper()
        return [f for f in FILINGS
                if q in f["debtor_name"].lower()
                and (not st or f["state"] == st)]
    if name == "get_filing_details":
        for f in FILINGS:
            if f["filing_number"] == args["filing_number"]:
                return f
        return {"error": "not found"}
    if name == "predict_delinquency":
        feats = [[args["active_filing_count"], args["state_count"],
                  args["collateral_types"], args["filing_age_years"],
                  args["amendment_frequency"], args["months_to_lapse"]]]
        prob = float(_MODEL.predict_proba(feats)[0][1])
        return {"probability": prob,
                "prediction": "HIGH" if prob > 0.66 else "MEDIUM" if prob > 0.33 else "LOW"}
    raise ValueError(f"Unknown tool: {name}")

"""agent.py — the raw API loop. Everything is hand-coded.

Note: we split the loop into a private _run_messages() helper that takes a
message list and returns (text, updated_messages). run_agent() is a thin
wrapper for the single-shot case; in Step 6 the session manager will call
_run_messages directly to support multi-turn."""
import json, anthropic
from tools import TOOLS, execute_tool

client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"
MAX_TURNS = 10
SYSTEM = """You are a UCC filing risk analyst. When given a business name, search
thoroughly using name variations including abbreviations and DBAs, gather filing
statistics, run the delinquency model, examine the riskiest filings, and write
a narrative report citing specific evidence."""

def _run_messages(messages: list) -> tuple[str, list]:
    """Drive the tool-use loop on the given messages list. Mutates messages.
    Returns (final_text, messages)."""
    for turn in range(MAX_TURNS):
        response = client.messages.create(
            model=MODEL,
            max_tokens=4096,
            system=SYSTEM,
            tools=TOOLS,
            messages=messages,
        )
        # Always append the assistant turn BEFORE handling tool calls.
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            text = next(b.text for b in response.content if b.type == "text")
            return text, messages

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    try:
                        result = execute_tool(block.name, block.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result, default=str),
                        })
                    except Exception as e:
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": f"ERROR: {e}",
                            "is_error": True,
                        })
            messages.append({"role": "user", "content": tool_results})
            continue

        raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason}")

    raise RuntimeError(f"Agent exceeded {MAX_TURNS} turns without finishing")

def run_agent(question: str) -> str:
    """Single-shot entry point. Wraps _run_messages with a fresh history."""
    text, _ = _run_messages([{"role": "user", "content": question}])
    return text

if __name__ == "__main__":
    print(run_agent("Assess the delinquency risk for Acme Corporation."))

"""guardrails.py — hand-coded checks called from the loop."""
import re

SSN_RE = re.compile(r"\b\d{3}-\d{2}-\d{4}\b")
PHONE_RE = re.compile(r"\b\d{3}-\d{3}-\d{4}\b")
COST_LIMIT_TOKENS = 50_000
CIRCUIT_FAIL_THRESHOLD = 3

def validate_input(tool_name: str, args: dict):
    if tool_name == "search_filings":
        q = args.get("debtor_name", "")
        if len(q.strip()) < 3:
            raise ValueError("Query too broad: debtor_name must be >=3 characters")

def redact_pii(payload: str) -> str:
    payload = SSN_RE.sub("[SSN_REDACTED]", payload)
    payload = PHONE_RE.sub("[PHONE_REDACTED]", payload)
    return payload

class CircuitBreaker:
    def __init__(self):
        self.failures = 0
    def record(self, ok: bool):
        if ok: self.failures = 0
        else:  self.failures += 1
        if self.failures >= CIRCUIT_FAIL_THRESHOLD:
            raise RuntimeError("Circuit breaker tripped — aborting")

# Add to the imports at the top of agent.py:
from guardrails import (validate_input, redact_pii,
                        CircuitBreaker, COST_LIMIT_TOKENS)   # NEW

_breaker = CircuitBreaker()        # NEW — module-level instance

def _run_messages(messages: list) -> tuple[str, list]:
    total_tokens = 0                # NEW — cost cap counter
    for turn in range(MAX_TURNS):
        response = client.messages.create(
            model=MODEL, max_tokens=4096, system=SYSTEM,
            tools=TOOLS, messages=messages,
        )
        total_tokens += (response.usage.input_tokens
                         + response.usage.output_tokens)        # NEW
        if total_tokens > COST_LIMIT_TOKENS:                    # NEW
            raise RuntimeError(f"Cost cap exceeded: {total_tokens} tokens")

        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            text = next(b.text for b in response.content if b.type == "text")
            return text, messages

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    try:
                        validate_input(block.name, block.input)  # NEW — pre-check
                        result = execute_tool(block.name, block.input)
                        _breaker.record(ok=True)                  # NEW
                        # Redact PII from the JSON we send back to Claude.
                        content = redact_pii(json.dumps(result, default=str))  # NEW
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": content,
                        })
                    except Exception as e:
                        _breaker.record(ok=False)                 # NEW — trips on 3rd fail
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": f"ERROR: {e}",
                            "is_error": True,
                        })
            messages.append({"role": "user", "content": tool_results})
            continue

        raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason}")

    raise RuntimeError(f"Agent exceeded {MAX_TURNS} turns without finishing")

python -c "from agent import run_agent; print(run_agent('Search for company A'))"
# Expected: agent attempts search_filings(debtor_name='A'), validate_input
# raises ValueError, the error is fed back to Claude as is_error: true,
# and Claude reformulates with a longer fragment.

"""audit.py — HIPAA-style audit log writer for the raw API loop."""
import json, datetime
from guardrails import redact_pii

def append_audit(tool_name: str, args: dict, result, tokens: int,
                 case_id: str = "default") -> None:
    rec = {
        "ts": datetime.datetime.now(datetime.timezone.utc)
                      .strftime("%Y-%m-%dT%H:%M:%SZ"),
        "case_id": case_id,
        "tool": tool_name,
        "input_redacted": redact_pii(json.dumps(args, default=str)),
        "output_summary": redact_pii(str(result))[:200],
        "tokens": tokens,
    }
    with open("audit_log.jsonl", "a") as f:
        f.write(json.dumps(rec) + "\n")

# Add to the imports at the top of agent.py:
from audit import append_audit                    # NEW

# Inside _run_messages, in the try-block right after `result = execute_tool(...)`:
                        result = execute_tool(block.name, block.input)
                        _breaker.record(ok=True)
                        append_audit(                              # NEW
                            tool_name=block.name,
                            args=block.input,
                            result=result,
                            tokens=total_tokens,
                            case_id=messages[0]["content"][:40] if messages else "default",
                        )
                        content = redact_pii(json.dumps(result, default=str))
                        # ... rest of the try-block unchanged ...

cat audit_log.jsonl   # macOS/Linux
type audit_log.jsonl  # Windows cmd
Get-Content audit_log.jsonl  # Windows PowerShell

"""session.py — multi-turn over the same _run_messages helper from Step 3."""
from agent import _run_messages

SESSIONS: dict[str, list] = {}   # session_id -> messages list
WINDOW = 20                      # keep the last N messages

def chat(session_id: str, user_msg: str) -> str:
    """Append the user's message to the session, run the loop, return the answer."""
    msgs = SESSIONS.setdefault(session_id, [])
    msgs.append({"role": "user", "content": user_msg})
    answer, msgs = _run_messages(msgs)
    # Sliding window: keep only the last WINDOW messages so context doesn't grow forever.
    SESSIONS[session_id] = msgs[-WINDOW:]
    return answer

python -c "
from session import chat
print(chat('case-001', 'What is the lien exposure for Acme Corporation?'))
print('---')
print(chat('case-001', 'Now compare to Pinnacle Industries.'))
"

"""server.py — FastAPI wrapper around run_agent + session.chat."""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent import run_agent
from session import chat as session_chat

app = FastAPI(title="UCC Risk Analyzer (Iter 1)")

class Q(BaseModel):
    question: str

class C(BaseModel):
    session_id: str
    message: str

@app.get("/health")
def health():
    return {"status": "ok", "iter": 1}

@app.post("/query")
def query(q: Q):
    try:
        return {"answer": run_agent(q.question)}
    except Exception as e:
        raise HTTPException(500, str(e))

@app.post("/chat")
def chat_ep(c: C):
    try:
        return {"answer": session_chat(c.session_id, c.message)}
    except Exception as e:
        raise HTTPException(500, str(e))

uvicorn server:app --reload --port 8000

curl localhost:8000/health
# Expected: {"status":"ok","iter":1}

curl -X POST localhost:8000/query \
     -H "Content-Type: application/json" \
     -d '{"question":"Risk for Acme Corporation?"}'
# Expected: {"answer":"Acme Corporation has 9 active UCC filings ..."}

docker build -t iter1-c .
docker run --rm -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY iter1-c
# In another terminal:
curl localhost:8000/health

mkdir agent-iter2-sdk && cd agent-iter2-sdk
python -m venv venv && source venv/bin/activate    # Windows: venv\Scripts\activate
pip install "claude-agent-sdk>=0.2" "scikit-learn>=1.3" "fastapi>=0.110" "uvicorn>=0.27"
npm i -g @anthropic-ai/claude-code   # if not already installed
claude
> /init

> Create agent.py using `claude-agent-sdk`. Define three UCC tools as
> @tool-decorated async functions: search_filings(debtor_name, state),
> get_filing_details(filing_number), predict_delinquency(...6 features).
> Wire them into a create_sdk_mcp_server, build ClaudeAgentOptions with the
> system prompt from CLAUDE.md, and expose an async run(question) entry
> point that drives query() and concatenates AssistantMessage text.

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "mcp__ucc__search_filings",
        "hooks": [
          { "type": "command", "command": "python hooks/log_call.py", "timeout": 5 }
        ]
      }
    ],
    "PostToolUse": [
      {
        "matcher": "*",
        "hooks": [
          { "type": "command", "command": "python hooks/redact_pii.py", "timeout": 10 },
          { "type": "command", "command": "python hooks/audit.py", "timeout": 10 }
        ]
      }
    ]
  }
}

"""hooks/log_call.py — PreToolUse: print the call to STDERR.

STDERR, deliberately. stdout is the channel a hook uses to return a DECISION,
so a log line there is parsed as a decision — and a malformed one. This is
the single most common way a working hook script breaks a working agent."""
import sys, json, datetime

payload = json.load(sys.stdin)
ts = (datetime.datetime.now(datetime.timezone.utc)
              .strftime("%Y-%m-%dT%H:%M:%SZ"))
name = payload.get("tool_name", "?")
args = payload.get("tool_input", {})
print(f"[{ts}] PRE  {name}({args})", file=sys.stderr)
# Nothing on stdout. This hook only observes, so it has no decision to make.

# Smoke-test the redactor. Note the key -- tool_respONSE is what a
# PostToolUse hook actually receives:
echo '{"tool_name":"search_filings","tool_response":"customer SSN 123-45-6789"}' \
  | python hooks/redact_pii.py
# Expect on stdout:
# {"hookSpecificOutput": {"hookEventName": "PostToolUse",
#  "updatedToolOutput": "customer SSN [SSN_REDACTED]"}}

# log_call writes to stderr and NOTHING to stdout. Check the second half --
# it is the part that silently breaks agents:
echo '{"tool_name":"search_filings","tool_input":{"debtor_name":"Acme"}}' \
  | python hooks/log_call.py | wc -c          # must print 0

# Smoke-test the audit writer (should append a line, and print nothing):
echo '{"tool_name":"search_filings","tool_input":{"debtor_name":"Acme"},"tool_response":"ok"}' \
  | python hooks/audit.py
cat audit_log.jsonl  # should show the new entry

python -c "import asyncio, agent; print(asyncio.run(agent.run('Risk for Acme Corporation?')))"

"""sessions.py — multi-turn via SDK resume tokens."""
from dataclasses import replace
from claude_agent_sdk import query, AssistantMessage
from agent import OPTIONS

SESSIONS: dict[str, str] = {}   # case_id -> resume token (session_id)

async def _drive(prompt: str, options) -> tuple[str, str | None]:
    parts, last_session_id = [], None
    async for msg in query(prompt=prompt, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if getattr(block, "text", None):
                    parts.append(block.text)
        # The SDK emits a session_id on the result message at the end.
        sid = getattr(msg, "session_id", None)
        if sid: last_session_id = sid
    return "\n".join(parts), last_session_id

async def chat(session_id: str, msg: str) -> str:
    resume = SESSIONS.get(session_id)
    options = replace(OPTIONS, resume=resume) if resume else OPTIONS
    text, sid = await _drive(msg, options)
    if sid: SESSIONS[session_id] = sid
    return text

async def what_if(session_id: str, hypothetical: str) -> str:
    """Fork: resume from the same point but do NOT update the main session."""
    resume = SESSIONS.get(session_id)
    options = replace(OPTIONS, resume=resume) if resume else OPTIONS
    text, _ = await _drive(hypothetical, options)
    return text

python -c "
import asyncio
from sessions import chat, what_if

async def main():
    # Turn 1: establish context
    print('T1:', await chat('case-001', 'What is the lien exposure for Acme Corporation?'))
    # Turn 2: builds on T1
    print('T2:', await chat('case-001', 'Now compare to Pinnacle Industries.'))
    # Fork: hypothetical that does NOT pollute the main session
    print('FORK:', await what_if('case-001', 'What if Acme files a UCC-3 termination on NY-2024-0848?'))
    # Turn 3 on the original session: the fork's hypothetical was NOT seen
    print('T3:', await chat('case-001', 'Summarize Acme vs Pinnacle in one sentence.'))

asyncio.run(main())
"

---
description: Run the UCC Risk Analyzer agent on a question
argument-hint: [question]
---
Run `python -c "import asyncio, agent; print(asyncio.run(agent.run('$ARGUMENTS')))"`.
If --verbose is in $ARGUMENTS, set AGENT_VERBOSE=1 (read by hooks/log_call.py).
Print the final report and the token + cost summary from query()'s ResultMessage.

claude
> /run-agent What is the lien exposure for Acme Corporation?
> /test-agent
> /eval-agent

> Create server.py and Dockerfile. Endpoints: GET /health, POST /query
> (single-shot), POST /chat (session_id + message). The endpoints are async
> FastAPI handlers that await agent.run() and sessions.chat() respectively
> (both are async coroutines from claude-agent-sdk). Dockerfile based on
> python:3.11-slim, install claude-agent-sdk + dependencies, expose 8000,
> uvicorn entrypoint. Mount .claude/ into the container so settings.json
> and hook scripts are picked up.

"""server.py — async FastAPI wrapper around the SDK agent."""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent import run as agent_run
from sessions import chat as session_chat

app = FastAPI(title="UCC Risk Analyzer (Iter 2 — SDK)")

class Q(BaseModel): question: str
class C(BaseModel): session_id: str; message: str

@app.get("/health")
def health(): return {"status": "ok", "iter": 2}

@app.post("/query")
async def query(q: Q):
    try: return {"answer": await agent_run(q.question)}
    except Exception as e: raise HTTPException(500, str(e))

@app.post("/chat")
async def chat_ep(c: C):
    try: return {"answer": await session_chat(c.session_id, c.message)}
    except Exception as e: raise HTTPException(500, str(e))

uvicorn server:app --reload --port 8000
# In another terminal:
curl localhost:8000/health
# Expected: {"status":"ok","iter":2}

curl -X POST localhost:8000/query -H "Content-Type: application/json" \
     -d '{"question":"Risk for Acme Corporation?"}'

docker build -t iter2-c .
docker run --rm -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY iter2-c

# Agent Specification: UCC Filing Risk Analyzer

## 1. Overview
An agent that assesses delinquency risk for businesses by searching UCC
filings, running an ML model, and writing a narrative risk report.

## 2. Configuration
- Model: claude-sonnet-4-6
- Framework: claude-agent-sdk (Python). Tools registered via
  create_sdk_mcp_server. Driver: query() with ClaudeAgentOptions.
- Max turns: 10
- Max tokens per response: 4096

## 3. Tools (registered as MCP server "ucc")

### search_filings  — mcp__ucc__search_filings
- @tool decorator with schema {debtor_name: str, state: str}
- Returns list of filing dicts wrapped in {"content": [{"type":"text","text": json}]}
- Mock data: 9 Acme filings across NY/CA/TX/FL incl. one DBA variant

### get_filing_details — mcp__ucc__get_filing_details
- {filing_number: str}

### predict_delinquency — mcp__ucc__predict_delinquency
- 6 features (active_filing_count, state_count, collateral_types,
  filing_age_years, amendment_frequency, months_to_lapse)
- Implementation: load sklearn RandomForest from delinquency_model.pkl

## 4. System Prompt (passed to ClaudeAgentOptions.system_prompt)
You are a UCC filing risk analyst. Search thoroughly using name variations
including abbreviations and DBAs, gather statistics, run the model, examine
riskiest filings, write a narrative report citing specific evidence.

## 5. Hooks
File-based via .claude/settings.json:
- PreToolUse  matcher "*"                command "python hooks/log_call.py"
- PostToolUse matcher "*"                command "python hooks/redact_pii.py"
- PostToolUse matcher "*"                command "python hooks/audit.py"

In-process via ClaudeAgentOptions.hooks (HookMatcher):
- PreToolUse matcher "mcp__ucc__search_filings": deny if debtor_name < 3 chars

PHI patterns to redact: SSN (NNN-NN-NNNN), phone (NNN-NNN-NNNN).

## 6. Sessions
Multi-turn via SDK resume tokens (sessions.py keeps a session_id -> resume map).
what_if() forks by re-using a resume token without persisting the new session_id.

## 7. Mock Data
9 Acme filings (NY:4, CA:2, TX:2, FL:1 DBA). 2 Pinnacle. 1 Sunrise.

## 8. API Wrapper
FastAPI async handlers: POST /query, POST /chat, GET /health.
Auth via X-API-Key header. Rate limit 10/min/key.

## 9. Deployment
Tier 1: Docker. Mount .claude/ into container so hooks resolve at runtime.

## 10. Tests
test_tools.py, test_agent.py (must find all 9 Acme incl. DBA),
test_hooks.py (validates HookMatcher denies short queries),
test_sessions.py (resume token reuse), test_api.py.

## 11. Evaluation
spec/test_scenarios.json with 10 questions covering name variations, state
filter, risk levels, what-if, broad synthesis, missing entities, lapse detection.

## 12. File Structure
ucc-risk-agent/
│── spec/agent-spec.md          # this file
│── spec/test_scenarios.json
│── CLAUDE.md
│── .claude/settings.json       # PreToolUse/PostToolUse matchers
│── .claude/commands/ {run-agent.md, test-agent.md, eval-agent.md}
│── agent.py                    # query() + @tool + create_sdk_mcp_server
│── sessions.py                 # resume-token sessions
│── mock_data.py | delinquency_model.pkl
│── hooks/{log_call,redact_pii,audit}.py
│── server.py | Dockerfile | docker-compose.yml | requirements.txt
│── tests/ x5
│── appendix/manual-loop.py     # Iter-1 reference, not for production

> /generate-from-spec spec/agent-spec.md

# (or, if /generate-from-spec is not installed:)
> Read spec/agent-spec.md and build the entire project. Create every file
> listed in section 12 (File Structure). Implement every tool, hook, test,
> and endpoint exactly as specified. Use the `claude-agent-sdk` package
> (NOT a fictional anthropic.Agent class). Tools must be @tool-decorated
> async functions registered via create_sdk_mcp_server. The driver is
> query() + ClaudeAgentOptions. Hooks live in .claude/settings.json
> AND in OPTIONS.hooks (HookMatcher) per section 5. Generate realistic
> mock UCC filing data. Train and save the pickle model. Also generate
> appendix/manual-loop.py as an under-the-hood reference.

# From inside Claude Code:
> /test-agent

# Or from a regular shell:
pytest tests/ -v

# Build and run via docker-compose (generated by Claude Code)
docker compose up --build -d
docker compose ps   # confirm container is "Up"

# Health check
curl localhost:8000/health
# Expected: {"status":"ok","iter":3}

# Same query as Iter 1 and Iter 2
curl -X POST localhost:8000/query \
     -H "Content-Type: application/json" \
     -d '{"question":"Risk for Acme Corporation?"}' | python -m json.tool

# Run the same query against all three deployments and diff
# (assumes Iter 1 on :8001, Iter 2 on :8002, Iter 3 on :8003)
for port in 8001 8002 8003; do
  echo "=== Iter on :$port ==="
  curl -s -X POST localhost:$port/query \
       -H "Content-Type: application/json" \
       -d '{"question":"Risk for Acme Corporation?"}' \
    | python -c "import json, sys; d = json.load(sys.stdin); print(d['answer'][:300])"
  echo
done