Patient Intake & Triage Agent

mkdir patient-intake && cd patient-intake
python3 -m venv venv
source venv/bin/activate
pip install anthropic python-dotenv requests

ANTHROPIC_API_KEY=sk-ant-...
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/...
CLINIC_NAME=Riverside Medical Practice
EMERGENCY_NUMBER=999

import os
import json
import uuid
import requests
from datetime import datetime
from dotenv import load_dotenv
import anthropic

load_dotenv()

claude       = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
CLINIC_NAME  = os.getenv("CLINIC_NAME", "Our Medical Practice")
EMERGENCY_NO = os.getenv("EMERGENCY_NUMBER", "999")

SYSTEM_PROMPT = f"""You are a patient intake assistant for {CLINIC_NAME}. 
Your role is to collect symptom information and determine triage urgency 
to help staff prepare for the patient's visit.

CRITICAL SAFETY RULE — ALWAYS FOLLOW:
If a patient describes ANY of these symptoms, IMMEDIATELY say:
"Based on what you've described, you may need emergency care. Please call 
{EMERGENCY_NO} (or go to A&E/ER) immediately. Do not wait."
Emergency symptoms: chest pain, difficulty breathing, stroke symptoms 
(sudden face drooping / arm weakness / speech difficulty), severe bleeding, 
unconsciousness or near-unconsciousness, suspected heart attack, severe 
head injury, anaphylaxis (severe allergic reaction).

INFORMATION TO COLLECT:
1. Patient's name and date of birth
2. Main symptom (what's the primary complaint?)
3. Duration (how long have they had this?)
4. Severity (on a scale of 1–10, with 10 being worst pain ever)
5. Associated symptoms (anything else they've noticed?)
6. Relevant medical history (any chronic conditions? Current medications?)
7. Preferred appointment time / urgency from their perspective

TRIAGE ASSESSMENT:
After collecting the information, assign one of:
- IMMEDIATE: Life-threatening (tell them to call {EMERGENCY_NO})
- VERY_URGENT: Needs to be seen within hours today
- URGENT: Should be seen same-day or next-day
- STANDARD: Routine appointment within 3 days
- NON_URGENT: Can wait up to 1 week

COMPLETION:
When you have all the information above (or the patient has indicated they're 
done), say exactly "INTAKE_COMPLETE" on its own line, then output the 
structured data as JSON on the next line.

JSON format:
{{"patient_name": "...", "dob": "...", "main_symptom": "...", 
  "duration": "...", "severity": "...", "associated_symptoms": "...",
  "medical_history": "...", "triage_level": "...", 
  "triage_reasoning": "...", "preferred_time": "..."}}

IMPORTANT RULES:
- Ask ONE question at a time — never a long list
- Be empathetic and professional
- Never diagnose — only collect and triage
- If uncertain about triage, err on the side of higher urgency
- Always remind patients that this is admin intake, not clinical advice"""

def alert_oncall(patient_data: dict, triage_level: str):
    """Send urgent alert to on-call staff via Slack."""
    urgency_emoji = {"IMMEDIATE": "🚨", "VERY_URGENT": "⚡"}.get(triage_level, "⏰")
    requests.post(os.getenv("SLACK_WEBHOOK_URL"), json={
        "blocks": [
            {"type": "header", "text": {"type": "plain_text",
             "text": f"{urgency_emoji} {triage_level} — Patient Intake Alert"}},
            {"type": "section", "fields": [
                {"type": "mrkdwn", "text": f"*Patient:*\n{patient_data.get('patient_name', 'Unknown')}"},
                {"type": "mrkdwn", "text": f"*DOB:*\n{patient_data.get('dob', 'Unknown')}"},
                {"type": "mrkdwn", "text": f"*Main Symptom:*\n{patient_data.get('main_symptom', 'Unknown')}"},
                {"type": "mrkdwn", "text": f"*Severity:*\n{patient_data.get('severity', 'Unknown')}/10"},
                {"type": "mrkdwn", "text": f"*Duration:*\n{patient_data.get('duration', 'Unknown')}"},
            ]},
            {"type": "section", "text": {"type": "mrkdwn",
             "text": f"*Triage reasoning:* {patient_data.get('triage_reasoning', 'See intake notes')}"}},
            {"type": "section", "text": {"type": "mrkdwn",
             "text": f"Intake logged at {datetime.now().strftime('%H:%M on %d %b %Y')}"}},
        ]
    })

def run_intake_agent():
    session_id   = str(uuid.uuid4())[:8].upper()
    conversation = []

    print("\n" + "="*60)
    print(f"Patient Intake — {CLINIC_NAME}")
    print(f"Session: {session_id}")
    print("="*60)
    print("(Type 'quit' to exit)\n")

    # Get initial greeting
    initial = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=400,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": "Hi, I need to see a doctor."}]
    )
    greeting = initial.content[0].text
    print(f"Intake Assistant: {greeting}\n")
    conversation.append({"role": "user", "content": "Hi, I need to see a doctor."})
    conversation.append({"role": "assistant", "content": greeting})

    while True:
        user_input = input("Patient: ").strip()
        if user_input.lower() == 'quit':
            break
        if not user_input:
            continue

        conversation.append({"role": "user", "content": user_input})

        response = claude.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=500,
            system=SYSTEM_PROMPT,
            messages=conversation
        )

        reply = response.content[0].text
        conversation.append({"role": "assistant", "content": reply})

        if "INTAKE_COMPLETE" in reply:
            # Find and parse JSON
            lines = reply.split('\n')
            for i, line in enumerate(lines):
                if line.strip() == 'INTAKE_COMPLETE' and i + 1 < len(lines):
                    try:
                        patient_data                  = json.loads(lines[i + 1])
                        patient_data['session_id']    = session_id
                        patient_data['intake_time']   = datetime.now().isoformat()
                        triage_level                  = patient_data.get('triage_level', 'STANDARD')

                        print(f"\nIntake Assistant: Thank you. Your intake is complete.")
                        print(f"\n--- INTAKE SUMMARY ---")
                        print(f"Triage Level : {triage_level}")
                        print(f"Main Symptom : {patient_data.get('main_symptom')}")
                        print(f"Severity     : {patient_data.get('severity')}/10")
                        print(f"Duration     : {patient_data.get('duration')}")
                        print(f"Reference    : INT-{session_id}")
                        print("----------------------")

                        if triage_level in ('IMMEDIATE', 'VERY_URGENT', 'URGENT'):
                            print(f"\n⚠️  Alerting on-call staff...")
                            alert_oncall(patient_data, triage_level)
                            print("✓ On-call staff notified")

                        print(f"\nPatient reference number: INT-{session_id}")
                        print("A member of staff will be in contact shortly.\n")

                    except (json.JSONDecodeError, IndexError) as e:
                        print(f"\nIntake complete. {reply.split('INTAKE_COMPLETE')[0].strip()}")
            break
        else:
            print(f"\nIntake Assistant: {reply}\n")

if __name__ == "__main__":
    run_intake_agent()

source venv/bin/activate
python intake_agent.py