M26 — Hooks, Sessions & the Agent SDK

mkdir m26-ucc-agent && cd m26-ucc-agent
python -m venv venv && source venv/bin/activate    # Windows: venv\Scripts\activate
pip install "claude-agent-sdk>=0.2"

# Optional: TypeScript variant
npm init -y && npm i @anthropic-ai/claude-agent-sdk zod tsx typescript

# Confirm the env var is set
echo $ANTHROPIC_API_KEY | head -c 12   # macOS/Linux  — should print your key prefix
echo %ANTHROPIC_API_KEY:~0,12%        # Windows cmd  — same idea
$env:ANTHROPIC_API_KEY.Substring(0,12) # Windows PowerShell

"""tools.py — 3 SDK tools registered through an in-process MCP server."""
import json
from claude_agent_sdk import tool, create_sdk_mcp_server
from mock_data import FILINGS_DB

@tool(
    "search_filings",
    "Search UCC filings by debtor name. Supports partial matching across states. "
    "Use this FIRST to find candidate filings before fetching details.",
    {"debtor_name": str, "state": str},
)
async def search_filings(args):
    name = args["debtor_name"].upper()
    state = args.get("state")
    hits = [
        f for f in FILINGS_DB
        if name in f["debtor_name"].upper()
           and (not state or f["state"] == state)
    ]
    return {"content": [{"type": "text", "text": json.dumps(hits)}]}

@tool(
    "get_filing_details",
    "Get full details for a specific UCC filing by filing number.",
    {"filing_number": str},
)
async def get_filing_details(args):
    for f in FILINGS_DB:
        if f["filing_number"] == args["filing_number"]:
            return {"content": [{"type": "text", "text": json.dumps(f)}]}
    return {"content": [{"type": "text",
                          "text": json.dumps({"error": "not found"})}]}

@tool(
    "calculate_risk_score",
    "Calculate a delinquency risk score for an entity by aggregating its filings.",
    {"entity_name": str},
)
async def calculate_risk_score(args):
    name = args["entity_name"].upper()
    matches = [f for f in FILINGS_DB if name in f["debtor_name"].upper()]
    active = [f for f in matches if f["status"] == "ACTIVE"]
    states = {f["state"] for f in active}
    score = min(0.95, 0.15 * len(active) + 0.05 * len(states))
    out = {"entity_name": args["entity_name"],
           "active_filings": len(active),
           "states": sorted(states),
           "risk_score": round(score, 2),
           "risk_level": "HIGH" if score > 0.66 else "MEDIUM" if score > 0.33 else "LOW"}
    return {"content": [{"type": "text", "text": json.dumps(out)}]}

ucc_server = create_sdk_mcp_server(
    name="ucc_tools",
    version="1.0.0",
    tools=[search_filings, get_filing_details, calculate_risk_score],
)

"""run_agent.py — the SDK entry point. ~25 lines, no manual loop."""
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage
from tools import ucc_server

OPTIONS = ClaudeAgentOptions(
    model="claude-sonnet-4-6",
    system_prompt=(
        "You are a UCC filing research agent. When given a business name, "
        "search thoroughly using name variations including abbreviations and "
        "DBAs. Then call calculate_risk_score and write a narrative report "
        "citing specific filings."
    ),
    mcp_servers={"ucc": ucc_server},
    allowed_tools=[
        "mcp__ucc__search_filings",
        "mcp__ucc__get_filing_details",
        "mcp__ucc__calculate_risk_score",
    ],
    max_turns=8,
)

async def main():
    final_text = ""
    async for msg in query(
        prompt="What is the lien exposure for Acme Corporation?",
        options=OPTIONS,
    ):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if hasattr(block, "text") and block.text:
                    final_text = block.text
    print(final_text)

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

# M15B agent_loop.py — manual tool-use loop (excerpt)
from anthropic import Anthropic
from tools import TOOL_SCHEMAS, dispatch_tool   # you wrote both

client = Anthropic()
messages = [{"role": "user", "content": "What is the lien exposure for Acme?"}]

for turn in range(8):                                    # max_turns
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        system="You are a UCC research agent...",
        tools=TOOL_SCHEMAS,                              # hand-written JSON Schema
        messages=messages,
        max_tokens=2048,
    )
    messages.append({"role": "assistant", "content": resp.content})

    if resp.stop_reason == "end_turn":                   # you check
        break
    if resp.stop_reason != "tool_use":
        raise RuntimeError(f"unexpected stop: {resp.stop_reason}")

    tool_results = []
    for block in resp.content:                           # you parse
        if block.type != "tool_use":
            continue
        try:
            result = dispatch_tool(block.name, block.input)   # you dispatch
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result),
            })
        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})  # you format

# Print the last assistant text block
final = next((b.text for b in resp.content if b.type == "text"), "")
print(final)

# Add these two imports to the top of run_agent.py:
from datetime import datetime
from claude_agent_sdk import HookMatcher

# Add this hook function above the OPTIONS definition:
async def log_tool_call(input_data, tool_use_id, context):
    ts = datetime.utcnow().isoformat() + "Z"
    name = input_data.get("tool_name")
    params = input_data.get("tool_input")
    print(f"[{ts}] PRE  {name}({params})")
    return {}   # empty dict = continue without modification

# Replace your existing OPTIONS = ClaudeAgentOptions(...) block with this version.
# Only the new `hooks=` kwarg is different from Step 3.
OPTIONS = ClaudeAgentOptions(
    model="claude-sonnet-4-6",
    system_prompt=(
        "You are a UCC filing research agent. When given a business name, "
        "search thoroughly using name variations including abbreviations and "
        "DBAs. Then call calculate_risk_score and write a narrative report "
        "citing specific filings."
    ),
    mcp_servers={"ucc": ucc_server},
    allowed_tools=[
        "mcp__ucc__search_filings",
        "mcp__ucc__get_filing_details",
        "mcp__ucc__calculate_risk_score",
    ],
    hooks={
        "PreToolUse": [HookMatcher(matcher="*", hooks=[log_tool_call])],
    },
    max_turns=8,
)

# Add to the top of run_agent.py:
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")

# Add this hook function above OPTIONS (next to log_tool_call):
async def redact_pii(input_data, tool_use_id, context):
    out = input_data.get("tool_response", "")
    if isinstance(out, str):
        out = SSN_RE.sub("[SSN REDACTED]", out)
        out = PHONE_RE.sub("[PHONE REDACTED]", out)
    return {"tool_response": out}

# Update the hooks={} dict inside your existing OPTIONS to add PostToolUse:
hooks={
    "PreToolUse":  [HookMatcher(matcher="*", hooks=[log_tool_call])],
    "PostToolUse": [HookMatcher(matcher="*", hooks=[redact_pii])],
},
# (leave the rest of OPTIONS — model, system_prompt, mcp_servers,
#  allowed_tools, max_turns — unchanged from Step 5)

# Add to the imports at the top of run_agent.py:
from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny

# Add this gate function above OPTIONS:
async def gate(tool_name, tool_input, context):
    # Block search_filings calls that would scan everything.
    if tool_name == "mcp__ucc__search_filings":
        q = (tool_input.get("debtor_name") or "").strip()
        if len(q) < 3:
            return PermissionResultDeny(
                message=f"Query too broad: '{q}' is < 3 chars. "
                        "Provide a longer name fragment."
            )
    return PermissionResultAllow()

# Add can_use_tool=gate to your existing OPTIONS block:
OPTIONS = ClaudeAgentOptions(
    model="claude-sonnet-4-6",
    system_prompt="...",                  # unchanged from Step 6
    mcp_servers={"ucc": ucc_server},
    allowed_tools=[                       # unchanged from Step 6
        "mcp__ucc__search_filings",
        "mcp__ucc__get_filing_details",
        "mcp__ucc__calculate_risk_score",
    ],
    hooks={                               # unchanged from Step 6
        "PreToolUse":  [HookMatcher(matcher="*", hooks=[log_tool_call])],
        "PostToolUse": [HookMatcher(matcher="*", hooks=[redact_pii])],
    },
    can_use_tool=gate,                    # <-- new in Step 7
    max_turns=8,
)

"""session_manager.py — thin multi-turn wrapper over query()."""
from claude_agent_sdk import query, AssistantMessage

class SessionManager:
    def __init__(self, options):
        self.options = options
        self.transcript: list[str] = []

    async def send(self, user_input: str) -> str:
        # Build a single prompt from prior turns plus the new input.
        prompt = "\n\n".join(self.transcript + [f"USER: {user_input}"])
        final = ""
        async for msg in query(prompt=prompt, options=self.options):
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if hasattr(block, "text") and block.text:
                        final = block.text
        # Persist the turn AFTER the agent has answered.
        self.transcript.append(f"USER: {user_input}")
        self.transcript.append(f"ASSISTANT: {final}")
        return final

    def fork(self) -> "SessionManager":
        clone = SessionManager(self.options)
        clone.transcript = list(self.transcript)   # deep enough for strings
        return clone

import asyncio
from session_manager import SessionManager
from run_agent import OPTIONS

async def main():
    s = SessionManager(OPTIONS)
    print(await s.send("What is the lien exposure for Acme Corporation?"))
    print(await s.send("Now compare to Pinnacle Industries."))

    branch = s.fork()
    # Hypothetical branch — does NOT contaminate `s`.
    print(await branch.send("What if Acme files a UCC-3 termination on NY-2024-0848?"))

    # Original session continues from turn 2.
    print(await s.send("Summarize the relative risk of Acme vs Pinnacle in one sentence."))

asyncio.run(main())

mkdir -p .claude/agents
# create the file with your editor of choice, contents shown in the next tab

# Replace the system_prompt= argument inside OPTIONS with this version:
system_prompt=(
    "You are the UCC research coordinator. For each user request:\n"
    " 1. Use search_filings (with name variations) and get_filing_details to "
    "gather all relevant filings.\n"
    " 2. Once gathered, delegate to the risk-analyst subagent for scoring and "
    "the one-paragraph profile.\n"
    " 3. Stitch the subagent's profile into your final answer along with "
    "specific filing citations.\n"
    "Do not call calculate_risk_score yourself — that's the subagent's job."
),

"""production_agent.py — the full M26 stack in one file.
Wires: @tool functions (incl. the prelude's pickled ML model) +
       create_sdk_mcp_server + ClaudeAgentOptions +
       PreToolUse/PostToolUse hooks + can_use_tool gate +
       SessionManager from Step 7."""
import json, pickle, re
from datetime import datetime
from claude_agent_sdk import (
    tool, create_sdk_mcp_server, query, ClaudeAgentOptions,
    AssistantMessage, HookMatcher,
    PermissionResultAllow, PermissionResultDeny,
)
from mock_data import FILINGS_DB
from session_manager import SessionManager     # from Step 7

# ---- Layer 1: tools (3 from earlier + 1 NEW ML tool from the prelude) -----
RISK_MODEL = pickle.load(open("models/risk_clf.pkl", "rb"))

@tool("search_filings",
      "Search UCC filings by debtor name. Supports partial matching.",
      {"debtor_name": str, "state": str})
async def search_filings(args):
    name, state = args["debtor_name"].upper(), args.get("state")
    hits = [f for f in FILINGS_DB
            if name in f["debtor_name"].upper()
               and (not state or f["state"] == state)]
    return {"content": [{"type": "text", "text": json.dumps(hits)}]}

@tool("get_filing_details",
      "Get full details for a specific UCC filing by filing number.",
      {"filing_number": str})
async def get_filing_details(args):
    rec = next((f for f in FILINGS_DB
                if f["filing_number"] == args["filing_number"]),
               {"error": "not found"})
    return {"content": [{"type": "text", "text": json.dumps(rec)}]}

@tool("ml_risk_predict",
      "Run the production RandomForest delinquency model on an entity. "
      "Returns probability + label. Use this INSTEAD of heuristic scoring.",
      {"entity_name": str})
async def ml_risk_predict(args):
    name = args["entity_name"].upper()
    matches = [f for f in FILINGS_DB if name in f["debtor_name"].upper()]
    active  = [f for f in matches if f["status"] == "ACTIVE"]
    states  = {f["state"] for f in active}
    secured = {f["secured_party"] for f in active}
    features = [[len(active), len(states), len(secured),
                 sum(1 for f in active if "UCC3" in f["filing_type"])]]
    proba = float(RISK_MODEL.predict_proba(features)[0][1])
    return {"content": [{"type": "text", "text": json.dumps({
        "entity_name": args["entity_name"],
        "probability_high_risk": round(proba, 3),
        "label": "HIGH" if proba > 0.66 else "MEDIUM" if proba > 0.33 else "LOW",
        "model": "RandomForest v1 (prelude)",
        "features_used": ["active_filings", "n_states",
                          "n_secured_parties", "n_continuations"],
    })}]}

ucc_server = create_sdk_mcp_server(
    name="ucc_tools", version="2.0.0",
    tools=[search_filings, get_filing_details, ml_risk_predict])

# ---- Layer 4: hooks + permission gate (Steps 5, 6, 7) ---------------------
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")

async def log_tool_call(input_data, tool_use_id, ctx):
    print(f"[{datetime.utcnow().isoformat()}Z] PRE  "
          f"{input_data.get('tool_name')}({input_data.get('tool_input')})")
    return {}

async def redact_pii(input_data, tool_use_id, ctx):
    out = input_data.get("tool_response", "")
    if isinstance(out, str):
        out = SSN_RE.sub("[SSN REDACTED]", out)
        out = PHONE_RE.sub("[PHONE REDACTED]", out)
    return {"tool_response": out}

async def gate(tool_name, tool_input, ctx):
    if tool_name == "mcp__ucc__search_filings":
        q = (tool_input.get("debtor_name") or "").strip()
        if len(q) < 3:
            return PermissionResultDeny(message=f"Query '{q}' too broad.")
    return PermissionResultAllow()

# ---- Layer 3: options ----------------------------------------------------
OPTIONS = ClaudeAgentOptions(
    model="claude-sonnet-4-6",
    system_prompt=(
        "You are the production UCC research coordinator. For each question:\n"
        " 1. Search filings using name variations (incl. abbreviations + DBAs).\n"
        " 2. Call ml_risk_predict for the canonical entity name — this is "
        "the production model; do not invent a heuristic.\n"
        " 3. Write a narrative report citing specific filings and the model's "
        "probability + label."
    ),
    mcp_servers={"ucc": ucc_server},
    allowed_tools=["mcp__ucc__search_filings",
                   "mcp__ucc__get_filing_details",
                   "mcp__ucc__ml_risk_predict"],
    hooks={"PreToolUse":  [HookMatcher(matcher="*", hooks=[log_tool_call])],
           "PostToolUse": [HookMatcher(matcher="*", hooks=[redact_pii])]},
    can_use_tool=gate,
    max_turns=10,
)

# ---- Layer 5: a session you can fork -------------------------------------
def new_session() -> SessionManager:
    return SessionManager(OPTIONS)