Capstone 7-A — Agent Evolution: Healthcare Pre-Auth

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

15 requests

"""tools.py — pre-auth tool schemas + executors for the raw API loop."""
import json
from pathlib import Path

DATA = Path("mock_data")
CRITERIA = json.loads((DATA / "clinical_criteria.json").read_text())
PROVIDERS = json.loads((DATA / "provider_directory.json").read_text())
BENEFITS = json.loads((DATA / "benefits.json").read_text())

TOOLS = [
    {
        "name": "lookup_clinical_criteria",
        "description": "Get medical-necessity criteria for a CPT procedure under a specific payer.",
        "input_schema": {
            "type": "object",
            "properties": {
                "cpt_code": {"type": "string", "pattern": "^[0-9]{5}$|^[A-Z][0-9]{4}$"},
                "payer": {"type": "string", "enum": ["Aetna", "UnitedHealth", "BlueCross"]},
            },
            "required": ["cpt_code", "payer"],
        },
    },
    {
        "name": "verify_diagnosis_match",
        "description": "Check if an ICD-10 code matches the approved indications in given criteria.",
        "input_schema": {
            "type": "object",
            "properties": {
                "icd10_code": {"type": "string"},
                "criteria_id": {"type": "string"},
            },
            "required": ["icd10_code", "criteria_id"],
        },
    },
    {
        "name": "check_network_status",
        "description": "Return the provider's in-network status for a given payer.",
        "input_schema": {
            "type": "object",
            "properties": {
                "provider_npi": {"type": "string", "pattern": "^[0-9]{10}$"},
                "payer": {"type": "string"},
            },
            "required": ["provider_npi", "payer"],
        },
    },
    {
        "name": "get_benefit_summary",
        "description": "Return coverage, copay, deductible-remaining for the member + procedure.",
        "input_schema": {
            "type": "object",
            "properties": {
                "member_id": {"type": "string"},
                "cpt_code": {"type": "string"},
            },
            "required": ["member_id", "cpt_code"],
        },
    },
    {
        "name": "generate_determination",
        "description": "Produce a final structured determination from gathered case data.",
        "input_schema": {
            "type": "object",
            "properties": {
                "case_id": {"type": "string"},
                "decision": {"type": "string", "enum": ["APPROVE", "DENY", "REQUEST_INFO"]},
                "rationale": {"type": "string"},
                "policy_citation": {"type": "string"},
            },
            "required": ["case_id", "decision", "rationale"],
        },
    },
]

def execute_tool(name, args):
    if name == "lookup_clinical_criteria":
        key = f"{args['cpt_code']}_{args['payer'].upper()}"
        return CRITERIA.get(key, {"error": "no policy found"})
    if name == "verify_diagnosis_match":
        # criteria_id is e.g. "27447_AETNA" or a policy_id
        for k, c in CRITERIA.items():
            if k == args["criteria_id"] or c.get("policy_id") == args["criteria_id"]:
                return {"match": args["icd10_code"] in c["approved_indications"],
                        "approved_indications": c["approved_indications"]}
        return {"error": "criteria not found"}
    if name == "check_network_status":
        p = PROVIDERS.get(args["provider_npi"])
        if not p: return {"error": "provider not found"}
        in_net = args["payer"] in p.get("networks", [])
        return {"npi": p["npi"], "specialty": p["specialty"],
                "in_network": in_net, "payer": args["payer"]}
    if name == "get_benefit_summary":
        key = f"{args['member_id']}_{args['cpt_code']}"
        return BENEFITS.get(key, {"covered": False, "error": "no benefit record"})
    if name == "generate_determination":
        return {"case_id": args["case_id"], "decision": args["decision"],
                "rationale": args["rationale"],
                "policy_citation": args.get("policy_citation", ""),
                "ts": "2025-04-15T12:00:00Z"}
    raise ValueError(f"Unknown tool: {name}")

"""agent.py — the raw API loop. Pre-auth decision flow.

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 case follow-ups."""
import json, anthropic
from tools import TOOLS, execute_tool

client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"
MAX_TURNS = 12
SYSTEM = """You are a pre-authorization decision agent for medical procedures.
For every case, follow this order:
1. lookup_clinical_criteria for the CPT + payer
2. verify_diagnosis_match for the patient's ICD-10
3. check_network_status for the provider + payer
4. get_benefit_summary for the member + CPT
5. generate_determination with APPROVE / DENY / REQUEST_INFO

Always cite the policy_id in your rationale. Never APPROVE without a successful
diagnosis match AND in-network provider AND covered benefit. Use REQUEST_INFO
when missing data, DENY when criteria fail."""

def _run_messages(messages: list) -> tuple[str, list]:
    """Drive the tool-use loop on the given messages list. 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 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__":
    q = ("Should pre-auth PA-2025-0001 (CPT 27447, ICD-10 M17.11, "
         "member A123456, provider NPI 1234567890, payer Aetna) be approved?")
    print(run_agent(q))

"""guardrails.py — HIPAA-aware checks for pre-auth agents."""
import re

# PHI patterns to redact from logs (NOT from the agent's working memory)
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")
MEMBER_ID_RE = re.compile(r"\b[A-Z]\d{6,9}\b")        # e.g. A123456
DOB_RE = re.compile(r"\b\d{4}-\d{2}-\d{2}\b")         # ISO date

CPT_RE = re.compile(r"^[0-9]{5}$|^[A-Z][0-9]{4}$")
NPI_RE = re.compile(r"^[0-9]{10}$")
ICD10_RE = re.compile(r"^[A-Z]\d{2}(\.\d{1,4})?$")

COST_LIMIT_TOKENS = 50_000
CIRCUIT_FAIL_THRESHOLD = 3

def validate_input(tool_name: str, args: dict):
    if tool_name == "lookup_clinical_criteria":
        if not CPT_RE.match(args.get("cpt_code", "")):
            raise ValueError(f"Invalid CPT code: {args.get('cpt_code')}")
    if tool_name == "verify_diagnosis_match":
        if not ICD10_RE.match(args.get("icd10_code", "")):
            raise ValueError(f"Invalid ICD-10 code: {args.get('icd10_code')}")
    if tool_name == "check_network_status":
        if not NPI_RE.match(args.get("provider_npi", "")):
            raise ValueError(f"Invalid NPI (must be 10 digits): {args.get('provider_npi')}")

def redact_phi(payload: str) -> str:
    """Scrub PHI from a string before writing to logs/audit."""
    payload = SSN_RE.sub("[SSN_REDACTED]", payload)
    payload = PHONE_RE.sub("[PHONE_REDACTED]", payload)
    payload = MEMBER_ID_RE.sub("[MEMBER_ID_REDACTED]", payload)
    payload = DOB_RE.sub("[DOB_REDACTED]", payload)
    return payload

class CircuitBreaker:
    def __init__(self): self.failures = 0
    def record(self, ok):
        self.failures = 0 if ok 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, 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 — CPT/NPI/ICD-10 check
                        result = execute_tool(block.name, block.input)
                        _breaker.record(ok=True)                        # NEW
                        # Pass the UNREDACTED result back to Claude.
                        # The agent needs the literal member_id to chain.
                        # PHI redaction happens inside append_audit (Step 5).
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result, default=str),
                        })
                    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('Pre-auth for CPT BAD123 ICD M17.11 NPI 12345 Aetna member A123456'))"
# Expected: agent attempts lookup_clinical_criteria(cpt_code='BAD123'),
# validate_input raises ValueError, error flows back as is_error: true,
# Claude either reformulates with a valid CPT or apologizes.

"""audit.py — HIPAA-compliant audit log writer."""
import json, datetime
from guardrails import redact_phi

def append_audit(tool_name: str, args: dict, result, tokens: int,
                 case_id: str = "default") -> None:
    rec = {
        "ts": datetime.datetime.utcnow().isoformat() + "Z",
        "case_id": case_id,
        "tool": tool_name,
        "input_redacted": redact_phi(json.dumps(args, default=str)),
        "output_summary": redact_phi(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)
                        # Use the case_id from the user's first message (PA-YYYY-NNNN).
                        case_id = next(
                            (w for w in messages[0]["content"].split() if w.startswith("PA-")),
                            "default"
                        )
                        append_audit(                              # NEW
                            tool_name=block.name,
                            args=block.input,
                            result=result,
                            tokens=total_tokens,
                            case_id=case_id,
                        )
                        tool_results.append({...})
                        # ... rest of 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 case sessions over the same _run_messages helper."""
from agent import _run_messages

SESSIONS: dict[str, list] = {}   # case_id -> messages list
WINDOW = 24                      # 5 tool calls per turn × ~5 turns + buffer

def chat(case_id: str, user_msg: str) -> str:
    """Append the user's message to the case session, run the loop, return the answer."""
    msgs = SESSIONS.setdefault(case_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[case_id] = msgs[-WINDOW:]
    return answer

python -c "
from session import chat
print(chat('PA-2025-0001', 'Should pre-auth PA-2025-0001 (CPT 27447, ICD M17.11, member A123456, NPI 1234567890, payer Aetna) be approved?'))
print('---')
print(chat('PA-2025-0001', 'What if the provider NPI were 9999999999 (out-of-network) instead?'))
"

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent import run_agent
from session import chat

app = FastAPI()
class Q(BaseModel): question: str
class C(BaseModel): case_id: str; message: str

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

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

@app.post("/chat")
def chat_ep(c: C):
    return {"answer": chat(c.case_id, c.message)}

anthropic>=0.40
fastapi>=0.110
uvicorn>=0.27
pydantic>=2.0

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

curl -X POST localhost:8000/preauth -H "Content-Type: application/json" \
     -d '{"question":"Is PA-2025-0001 approvable?"}'
# Expected: {"determination":"Determination: APPROVE..."}

docker build -t iter1-a .
docker run --rm -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY iter1-a

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" "fastapi>=0.110" "uvicorn>=0.27" "pydantic>=2.0"
npm i -g @anthropic-ai/claude-code   # if not already installed
claude
> /init

> Create agent.py using `claude-agent-sdk`. Define five pre-auth tools as
> @tool-decorated async functions returning MCP-shaped {"content":[...]}:
> lookup_clinical_criteria, verify_diagnosis_match, check_network_status,
> get_benefit_summary, generate_determination. Wire them into a
> create_sdk_mcp_server, build ClaudeAgentOptions with the system prompt
> from CLAUDE.md, and expose async run(question) driving query() and
> concatenating AssistantMessage text.

{
  "hooks": {
    "PreToolUse": [
      { "matcher": "*", "command": "python hooks/log_redacted.py" }
    ],
    "PostToolUse": [
      { "matcher": "*", "command": "python hooks/audit_redacted.py" }
    ]
  }
}

"""hooks/log_redacted.py — PreToolUse: print redacted call info to stderr."""
import sys, json, re, datetime

MEMBER_RE = re.compile(r"\b[A-Z]\d{6,9}\b")
SSN_RE    = re.compile(r"\b\d{3}-\d{2}-\d{4}\b")

payload = json.load(sys.stdin)
ts = datetime.datetime.utcnow().isoformat() + "Z"
name = payload.get("tool_name", "?")
args_str = json.dumps(payload.get("tool_input", {}))
args_str = MEMBER_RE.sub("[MEMBER_ID_REDACTED]", args_str)
args_str = SSN_RE.sub("[SSN_REDACTED]", args_str)
print(f"[{ts}] PRE  {name}({args_str})", file=sys.stderr)
json.dump(payload, sys.stdout)   # CRITICAL: pass-through unchanged

# Test the log hook (should print redacted to stderr, pass through stdout):
echo '{"tool_name":"get_benefit_summary","tool_input":{"member_id":"A123456","cpt_code":"27447"}}' \
  | python hooks/log_redacted.py

# Test the audit hook (should append a redacted line to audit_log.jsonl):
echo '{"tool_name":"get_benefit_summary","tool_input":{"member_id":"A123456"},"tool_result":{"covered":true}}' \
  | python hooks/audit_redacted.py
cat audit_log.jsonl

python -c "import asyncio, agent; print(asyncio.run(agent.run('Is PA-2025-0001 approvable? CPT 27447 ICD M17.11 member A123456 NPI 1234567890 Aetna')))"

"""sessions.py — multi-case 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, options):
    parts, sid = [], 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)
        s = getattr(msg, "session_id", None)
        if s: sid = s
    return "\n".join(parts), sid

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

async def what_if(case_id: str, hypothetical: str) -> str:
    """Fork: 'what if the payer were UnitedHealth?' — do NOT save the new sid."""
    resume = SESSIONS.get(case_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():
    print('T1:', await chat('PA-2025-0001', 'Is this approvable? CPT 27447, ICD M17.11, member A123456, NPI 1234567890, Aetna'))
    print('FORK:', await what_if('PA-2025-0001', 'What if the payer were UnitedHealth instead?'))
    print('T2:', await chat('PA-2025-0001', 'Stick with Aetna. What about a peer-to-peer review?'))

asyncio.run(main())
"

---
description: Run the pre-auth agent on a case_id from preauth_requests.json
argument-hint: [case_id]
---
Look up case $ARGUMENTS in mock_data/preauth_requests.json. Build the question
string. Run `python -c "import asyncio, agent; print(asyncio.run(agent.run(q)))"`
where q is the question. Print determination, rationale, policy citation,
total tokens, total cost from the ResultMessage emitted by query().

> Create server.py and Dockerfile. Endpoints: GET /health,
> POST /preauth (single-shot determination), POST /chat (case_id + message).
> Async FastAPI handlers awaiting agent.run() and sessions.chat() (both are
> async coroutines from claude-agent-sdk). python:3.11-slim base, install
> claude-agent-sdk + dependencies, expose 8000. Mount .claude/ into the
> container so settings.json + hook scripts resolve at runtime.

uvicorn server:app --reload --port 8000
# In another terminal:
curl localhost:8000/health
curl -X POST localhost:8000/preauth -H "Content-Type: application/json" \
     -d '{"question":"Is PA-2025-0001 approvable?"}'

# Then build the container (mounts .claude/ at COPY time):
docker build -t iter2-a .
docker run --rm -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY iter2-a