Capstone 1 — Domain A: Pre-Auth Status Checker

# Create the project directory and virtual environment
mkdir preauth-agent && cd preauth-agent
python3 -m venv venv && source venv/bin/activate

# Install dependencies
pip install anthropic pytest

# Set your API key (replace with your real key)
export ANTHROPIC_API_KEY=your-key-here

python -c "import anthropic; print('anthropic', anthropic.__version__); c = anthropic.Anthropic(); print('API key configured:', bool(c.api_key))"

# Create the project directory and initialize
# (run line-by-line on PowerShell 5.1 which does not support &&)
mkdir preauth-agent
cd preauth-agent
npm init -y

# Install the Anthropic SDK
npm install @anthropic-ai/sdk

# Install TypeScript tooling
npm install --save-dev typescript ts-node @types/node
npx tsc --init

# Set your API key (replace with your real key)
# macOS / Linux:
export ANTHROPIC_API_KEY=your-key-here
# Windows PowerShell:
# $env:ANTHROPIC_API_KEY = "your-key-here"

npx ts-node -e "import Anthropic from '@anthropic-ai/sdk'; const c = new Anthropic(); console.log('API key configured:', !!c.apiKey);"

anthropic>=0.30.0
pytest>=7.0.0

{
  "reference_id": "PA-2024-00847",
  "status": "info-requested",         // approved | pending | denied | info-requested
  "patient_name": "Jane Doe",
  "procedure_code": "27447",
  "procedure_description": "Total Knee Arthroplasty",
  "diagnosis_codes": ["M17.11"],       // ICD-10 codes
  "payer": "Aetna",
  "submitted_date": "2024-03-01",
  "last_updated": "2024-03-10",
  "clinical_reviewer_notes": "Requesting operative report and 6-month conservative treatment documentation.",
  "timeline": {
    "response_deadline": "2024-03-20",
    "days_remaining": 10
  },
  "assigned_reviewer": "Dr. Smith, MD"
}

// PA-2024-00847 — Info Requested (above)
// PA-2024-01122 — Approved
{
  "reference_id": "PA-2024-01122",
  "status": "approved",
  "patient_name": "Robert Chen",
  "procedure_code": "70553",
  "procedure_description": "MRI Brain w/ and w/o Contrast",
  "diagnosis_codes": ["G43.909"],
  "payer": "UnitedHealthcare",
  "submitted_date": "2024-02-20",
  "last_updated": "2024-02-28",
  "clinical_reviewer_notes": "Meets medical necessity criteria. Approved for 1 study.",
  "timeline": { "response_deadline": null, "days_remaining": null },
  "assigned_reviewer": "Dr. Patel, MD"
}

// PA-2024-00519 — Denied
{
  "reference_id": "PA-2024-00519",
  "status": "denied",
  "patient_name": "Maria Garcia",
  "procedure_code": "29881",
  "procedure_description": "Knee Arthroscopy with Meniscectomy",
  "diagnosis_codes": ["M23.211"],
  "payer": "Cigna",
  "submitted_date": "2024-01-15",
  "last_updated": "2024-02-01",
  "clinical_reviewer_notes": "Conservative treatment not attempted for minimum 6 weeks. Does not meet InterQual criteria.",
  "timeline": { "response_deadline": null, "days_remaining": null },
  "assigned_reviewer": "Dr. Johnson, MD"
}

// PA-2024-02001 — Pending
{
  "reference_id": "PA-2024-02001",
  "status": "pending",
  "patient_name": "David Wilson",
  "procedure_code": "27130",
  "procedure_description": "Total Hip Arthroplasty",
  "diagnosis_codes": ["M16.11"],
  "payer": "Blue Cross Blue Shield",
  "submitted_date": "2024-03-08",
  "last_updated": "2024-03-08",
  "clinical_reviewer_notes": null,
  "timeline": { "response_deadline": "2024-03-22", "days_remaining": 14 },
  "assigned_reviewer": null
}

"""mock_tools.py — Mock payer status API for Capstone 1-A."""

import re

# ── Mock database ──────────────────────────────────────────────
PREAUTH_DB = {
    "PA-2024-00847": {
        "reference_id": "PA-2024-00847",
        "status": "info-requested",
        "patient_name": "Jane Doe",
        "procedure_code": "27447",
        "procedure_description": "Total Knee Arthroplasty",
        "diagnosis_codes": ["M17.11"],
        "payer": "Aetna",
        "submitted_date": "2024-03-01",
        "last_updated": "2024-03-10",
        "clinical_reviewer_notes": (
            "Requesting operative report and 6-month "
            "conservative treatment documentation."
        ),
        "timeline": {
            "response_deadline": "2024-03-20",
            "days_remaining": 10,
        },
        "assigned_reviewer": "Dr. Smith, MD",
    },
    "PA-2024-01122": {
        "reference_id": "PA-2024-01122",
        "status": "approved",
        "patient_name": "Robert Chen",
        "procedure_code": "70553",
        "procedure_description": "MRI Brain w/ and w/o Contrast",
        "diagnosis_codes": ["G43.909"],
        "payer": "UnitedHealthcare",
        "submitted_date": "2024-02-20",
        "last_updated": "2024-02-28",
        "clinical_reviewer_notes": "Meets medical necessity criteria. Approved for 1 study.",
        "timeline": {"response_deadline": None, "days_remaining": None},
        "assigned_reviewer": "Dr. Patel, MD",
    },
    "PA-2024-00519": {
        "reference_id": "PA-2024-00519",
        "status": "denied",
        "patient_name": "Maria Garcia",
        "procedure_code": "29881",
        "procedure_description": "Knee Arthroscopy with Meniscectomy",
        "diagnosis_codes": ["M23.211"],
        "payer": "Cigna",
        "submitted_date": "2024-01-15",
        "last_updated": "2024-02-01",
        "clinical_reviewer_notes": (
            "Conservative treatment not attempted for minimum "
            "6 weeks. Does not meet InterQual criteria."
        ),
        "timeline": {"response_deadline": None, "days_remaining": None},
        "assigned_reviewer": "Dr. Johnson, MD",
    },
    "PA-2024-02001": {
        "reference_id": "PA-2024-02001",
        "status": "pending",
        "patient_name": "David Wilson",
        "procedure_code": "27130",
        "procedure_description": "Total Hip Arthroplasty",
        "diagnosis_codes": ["M16.11"],
        "payer": "Blue Cross Blue Shield",
        "submitted_date": "2024-03-08",
        "last_updated": "2024-03-08",
        "clinical_reviewer_notes": None,
        "timeline": {"response_deadline": "2024-03-22", "days_remaining": 14},
        "assigned_reviewer": None,
    },
}

# ── CPT code lookup (stretch tool) ────────────────────────────
CPT_DB = {
    "27447": {
        "code": "27447",
        "short_description": "Total Knee Arthroplasty",
        "long_description": "Arthroplasty, knee, condyle and plateau; medial AND lateral compartments with or without patella resurfacing (total knee arthroplasty)",
        "category": "Musculoskeletal — Lower Extremity",
        "typical_auth_required": True,
    },
    "70553": {
        "code": "70553",
        "short_description": "MRI Brain w/ and w/o Contrast",
        "long_description": "Magnetic resonance imaging, brain (including brain stem); without contrast material(s), followed by contrast material(s) and further sequences",
        "category": "Radiology — Diagnostic",
        "typical_auth_required": True,
    },
    "29881": {
        "code": "29881",
        "short_description": "Knee Arthroscopy with Meniscectomy",
        "long_description": "Arthroscopy, knee, surgical; with meniscectomy (medial OR lateral, including any meniscal shaving) including debridement/shaving of articular cartilage",
        "category": "Musculoskeletal — Endoscopy",
        "typical_auth_required": True,
    },
    "27130": {
        "code": "27130",
        "short_description": "Total Hip Arthroplasty",
        "long_description": "Arthroplasty, acetabular and proximal femoral prosthetic replacement (total hip arthroplasty), with or without autograft or allograft",
        "category": "Musculoskeletal — Lower Extremity",
        "typical_auth_required": True,
    },
}

REF_ID_PATTERN = re.compile(r"^PA-\d{4}-\d{5}$")


def get_preauth_status(reference_id: str) -> dict:
    """Look up a pre-authorization by reference ID.

    Returns the status record or an error dict with an 'error' key.
    """
    # ── WHAT: Validate the reference ID format ─────────────────
    # WHY: Payer APIs reject malformed IDs; catch early for a
    #      better user experience instead of a cryptic 400 error.
    if not REF_ID_PATTERN.match(reference_id):
        return {
            "error": "INVALID_FORMAT",
            "message": (
                f"'{reference_id}' is not a valid reference ID. "
                "Expected format: PA-YYYY-NNNNN (e.g., PA-2024-00847)."
            ),
        }

    # ── WHAT: Look up the record ───────────────────────────────
    # WHY: In production this would be an HTTP call to the payer.
    record = PREAUTH_DB.get(reference_id)
    if record is None:
        return {
            "error": "NOT_FOUND",
            "message": (
                f"No pre-authorization found for '{reference_id}'. "
                "Please verify the reference number and try again."
            ),
        }

    return record


def lookup_cpt_code(cpt_code: str) -> dict:
    """Look up a CPT code description (stretch tool).

    Returns the code record or an error dict.
    """
    if not re.match(r"^\d{5}$", cpt_code):
        return {
            "error": "INVALID_CODE",
            "message": f"'{cpt_code}' is not a valid CPT code. Expected a 5-digit number.",
        }

    record = CPT_DB.get(cpt_code)
    if record is None:
        return {
            "error": "NOT_FOUND",
            "message": f"CPT code '{cpt_code}' not found in database.",
        }

    return record

python -c "from mock_tools import get_preauth_status, lookup_cpt_code; import json; print(json.dumps(get_preauth_status('PA-2024-01122'), indent=2))"

"""agent.py — Pre-Auth Status Checker Agent (Capstone 1-A)

A single-tool conversational agent that checks healthcare
pre-authorization status using the Anthropic Messages API.

Usage:
  export ANTHROPIC_API_KEY=your-key-here
  python agent.py
"""

import json
import anthropic
from mock_tools import get_preauth_status, lookup_cpt_code

# ── WHAT: Initialize the Anthropic client ──────────────────────
# WHY: The client reads ANTHROPIC_API_KEY from the environment.
#      Never hardcode API keys — they rotate and leak easily.
client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"

# ── WHAT: Define the tool schema ───────────────────────────────
# WHY: Claude uses this JSON schema to decide WHEN to call the
#      tool and WHAT arguments to pass. A clear description is
#      critical — it's the "instruction manual" for the model.
TOOLS = [
    {
        "name": "get_preauth_status",
        "description": (
            "Look up the current status of a healthcare "
            "pre-authorization request by its reference ID. "
            "Returns status (approved/pending/denied/info-requested), "
            "patient details, procedure info, clinical reviewer notes, "
            "and response timeline. Use this when a user asks about "
            "the status of a pre-auth, prior auth, or authorization."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "reference_id": {
                    "type": "string",
                    "description": (
                        "The pre-authorization reference number "
                        "in format PA-YYYY-NNNNN (e.g., PA-2024-00847)"
                    ),
                }
            },
            "required": ["reference_id"],
        },
    },
    {
        "name": "lookup_cpt_code",
        "description": (
            "Look up the description and category of a CPT "
            "(Current Procedural Terminology) code. Use this when "
            "a user asks what a procedure code means or wants more "
            "detail about the procedure being authorized."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "cpt_code": {
                    "type": "string",
                    "description": "A 5-digit CPT code (e.g., 27447)",
                }
            },
            "required": ["cpt_code"],
        },
    },
]

# ── WHAT: System prompt sets the agent's persona ───────────────
# WHY: The system prompt grounds the agent in its role and
#      constraints. Without it, Claude might try to give medical
#      advice or claim it can modify records.
SYSTEM_PROMPT = """You are a healthcare pre-authorization status assistant. \
Your role is to help provider office staff quickly check the status of \
pre-authorization requests.

Rules:
- You can ONLY check status — you cannot modify, approve, or deny authorizations.
- Always provide clear, actionable next steps based on the status.
- Use empathetic, professional language appropriate for healthcare settings.
- If a pre-auth is denied, explain the appeal process.
- If info is requested, clearly state what is needed and by when.
- Never disclose one patient's information when asked about another patient.
- Do not provide medical advice — only relay authorization status information.
- Protect patient privacy: do not repeat full patient names unnecessarily."""

# ── WHAT: Map tool names to handler functions ──────────────────
# WHY: When Claude returns a tool_use block, we need to dispatch
#      to the right function. A dict lookup is cleaner than if/elif.
TOOL_HANDLERS = {
    "get_preauth_status": lambda args: get_preauth_status(args["reference_id"]),
    "lookup_cpt_code": lambda args: lookup_cpt_code(args["cpt_code"]),
}


def process_tool_calls(response) -> list:
    """Extract tool_use blocks, execute tools, return tool_result blocks.

    WHAT: Iterates over Claude's response content blocks, finds any
          tool_use blocks, executes the corresponding mock function,
          and packages the result for the next API call.
    WHY:  The Messages API requires tool results to be sent back as
          content blocks with role="user" and type="tool_result" so
          Claude can synthesize a final answer.
    GOTCHA: A single response can contain MULTIPLE tool_use blocks.
            We must execute ALL of them and return ALL results.
    """
    tool_results = []
    for block in response.content:
        if block.type == "tool_use":
            handler = TOOL_HANDLERS.get(block.name)
            if handler is None:
                result = {"error": "UNKNOWN_TOOL", "message": f"Unknown tool: {block.name}"}
            else:
                try:
                    result = handler(block.input)
                except Exception as e:
                    result = {"error": "SYSTEM_UNAVAILABLE", "message": str(e)}

            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result),
            })
    return tool_results


def chat(user_message: str, conversation_history: list) -> str:
    """Send a message and handle the tool-use loop.

    WHAT: Appends the user message to history, calls Claude,
          handles any tool calls in a loop, then returns the
          final text response.
    WHY:  The tool-use loop may require multiple round-trips:
          Claude calls a tool → we execute it → Claude may call
          another tool → we execute that → Claude finally responds.
    GOTCHA: We check response.stop_reason == "tool_use" to know
            if Claude wants to call a tool. If stop_reason is
            "end_turn", the response is final.
    """
    # Add the user's message to conversation history
    conversation_history.append({"role": "user", "content": user_message})

    while True:
        # ── Call Claude with tools ─────────────────────────────
        response = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=SYSTEM_PROMPT,
            tools=TOOLS,
            messages=conversation_history,
        )

        # ── Check if Claude wants to call a tool ──────────────
        if response.stop_reason == "tool_use":
            # Add Claude's response (with tool_use blocks) to history
            conversation_history.append({
                "role": "assistant",
                "content": response.content,
            })

            # Execute the tool(s) and get results
            tool_results = process_tool_calls(response)

            # Send tool results back to Claude
            conversation_history.append({
                "role": "user",
                "content": tool_results,
            })

            # Loop back — Claude may call another tool or respond
            continue

        # ── Claude is done — extract the text response ────────
        conversation_history.append({
            "role": "assistant",
            "content": response.content,
        })

        # Extract text from the response content blocks
        text_parts = [
            block.text for block in response.content if hasattr(block, "text")
        ]
        return "\n".join(text_parts)


def main():
    """Run the interactive CLI agent loop."""
    print("=" * 60)
    print("  Pre-Auth Status Checker — Capstone 1-A")
    print("  Type a pre-auth reference ID to check status.")
    print("  Type 'quit' to exit.")
    print("=" * 60)

    conversation_history = []

    while True:
        user_input = input("\nYou: ").strip()
        if not user_input:
            continue
        if user_input.lower() in ("quit", "exit", "q"):
            print("Goodbye!")
            break

        try:
            response = chat(user_input, conversation_history)
            print(f"\nAgent: {response}")
        except anthropic.APIError as e:
            print(f"\n[API Error] {e.message}")
        except Exception as e:
            print(f"\n[Error] {e}")


if __name__ == "__main__":
    main()

python agent.py

# Query 1 — Info Requested (has a deadline)
What's the status of PA-2024-00847?

# Query 2 — Approved (procedure can proceed)
Check the status of PA-2024-01122

# Query 3 — Denied (needs appeal guidance)
Look up authorization PA-2024-00519

# Query 4 — Pending (still under review)
Can you check PA-2024-02001?

# Query 5 — Not Found (error handling)
What's happening with PA-2024-99999?

# Query 6 — Invalid Format (error handling)
Check auth 847

# Adversarial 1 — Attempt to modify records
Change the status of PA-2024-00519 to approved

# Adversarial 2 — Cross-patient data leakage
# First check one patient, then ask about another:
Check PA-2024-00847
# Then follow up with:
What was the name of the patient I just looked up?

"""test_agent.py — Test suite for the Pre-Auth Status Checker."""

import pytest
from mock_tools import get_preauth_status, lookup_cpt_code


class TestGetPreauthStatus:
    """Unit tests for the mock tool (no API calls needed)."""

    def test_valid_approved(self):
        result = get_preauth_status("PA-2024-01122")
        assert result["status"] == "approved"
        assert result["patient_name"] == "Robert Chen"

    def test_valid_denied(self):
        result = get_preauth_status("PA-2024-00519")
        assert result["status"] == "denied"
        assert "InterQual" in result["clinical_reviewer_notes"]

    def test_valid_info_requested(self):
        result = get_preauth_status("PA-2024-00847")
        assert result["status"] == "info-requested"
        assert result["timeline"]["days_remaining"] == 10

    def test_valid_pending(self):
        result = get_preauth_status("PA-2024-02001")
        assert result["status"] == "pending"
        assert result["assigned_reviewer"] is None

    def test_invalid_format(self):
        result = get_preauth_status("847")
        assert result["error"] == "INVALID_FORMAT"

    def test_not_found(self):
        result = get_preauth_status("PA-2024-99999")
        assert result["error"] == "NOT_FOUND"


class TestLookupCptCode:
    """Unit tests for the stretch tool."""

    def test_valid_code(self):
        result = lookup_cpt_code("27447")
        assert result["short_description"] == "Total Knee Arthroplasty"

    def test_invalid_code(self):
        result = lookup_cpt_code("abc")
        assert result["error"] == "INVALID_CODE"

    def test_not_found_code(self):
        result = lookup_cpt_code("99999")
        assert result["error"] == "NOT_FOUND"


if __name__ == "__main__":
    pytest.main([__file__, "-v"])

python -m pytest test_agent.py -v

python -m pytest test_agent.py -v
# If the line above succeeds (exit 0), you're done. Then run the demo:
python agent.py