#!/usr/bin/env python3
"""
Comprehensive Examples and Tests for MCP Healthcare System
This file demonstrates:
1. Tool execution examples
2. Clinical scenario simulations
3. Data model usage
4. FHIR resource creation
5. Integration testing
"""
import json
import asyncio
from datetime import datetime, timedelta
from typing import List, Dict, Any
# Import from our modules
from mcp_healthcare_server import (
get_patient_vitals,
get_patient_labs,
get_patient_medications,
check_drug_interactions,
order_medication,
generate_discharge_summary,
ehr_client,
audit_logger
)
from healthcare_utils import (
FHIRObservation,
FHIRPatient,
PIIRedactor,
ClinicalScores,
MedicationDatabase,
HashUtils
)
from config import get_config
# ============================================================================
# EXAMPLE 1: Patient Data Retrieval
# ============================================================================
async def example_patient_retrieval():
"""Example: Retrieve patient data"""
print("\n" + "="*70)
print("EXAMPLE 1: Patient Data Retrieval")
print("="*70)
context = {
"user_id": "doc-001",
"user_role": "physician",
"ip_address": "192.168.1.100"
}
params = {"patient_id": "MRN001", "limit": 5}
print(f"\nFetching vitals for patient MRN001...")
result = await get_patient_vitals(params, context)
if result.get("success"):
print(f"✓ Retrieved {result['count']} vital readings")
for vital in result.get("vitals", [])[:2]:
code = vital.get("code", {})
value = vital.get("valueQuantity", {})
print(f" - {code.get('text')}: {value.get('value')} {value.get('unit')}")
else:
print(f"✗ Error: {result.get('error')}")
# ============================================================================
# EXAMPLE 2: Lab Interpretation
# ============================================================================
async def example_lab_interpretation():
"""Example: Retrieve and interpret labs"""
print("\n" + "="*70)
print("EXAMPLE 2: Lab Results with Interpretation")
print("="*70)
context = {
"user_id": "doc-001",
"user_role": "physician"
}
params = {"patient_id": "MRN001"}
result = await get_patient_labs(params, context)
if result.get("success"):
print(f"\nLab Results for patient:")
for lab in result.get("labs", []):
code = lab.get("code", {})
value = lab.get("valueQuantity", {})
ref_range = lab.get("referenceRange", [{}])[0]
interpretation = lab.get("interpretation", [{}])[0]
interp_code = interpretation.get("coding", [{}])[0].get("code")
interp_text = {"N": "Normal", "H": "High", "L": "Low"}.get(interp_code, "Unknown")
print(f"\n Lab: {code.get('text')}")
print(f" Value: {value.get('value')} {value.get('unit')}")
print(f" Reference: {ref_range.get('low', {}).get('value')}-{ref_range.get('high', {}).get('value')}")
print(f" Status: {interp_text} {'⚠️' if interp_code != 'N' else '✓'}")
# ============================================================================
# EXAMPLE 3: Medication Management
# ============================================================================
async def example_medication_management():
"""Example: Get medications and check interactions"""
print("\n" + "="*70)
print("EXAMPLE 3: Medication Management & Drug Interactions")
print("="*70)
context = {
"user_id": "doc-001",
"user_role": "physician"
}
# Get current medications
print("\nFetching current medications...")
result = await get_patient_medications({"patient_id": "MRN001"}, context)
if result.get("success"):
meds = result.get("medications", [])
print(f"✓ Patient is on {len(meds)} medications:")
drug_names = []
for med in meds:
print(f" - {med['drug_name']} {med['dose']}{med['unit']} {med['frequency']}")
drug_names.append(med["drug_name"])
# Check for interactions
if len(drug_names) > 1:
print(f"\nChecking for drug interactions...")
interaction_result = await check_drug_interactions(
{"drugs": drug_names},
context
)
interactions = interaction_result.get("interactions", [])
if interactions:
print(f"⚠️ Found {len(interactions)} interaction(s):")
for interaction in interactions:
print(f"\n {interaction['drug1']} + {interaction['drug2']}")
print(f" Severity: {interaction['severity'].upper()}")
print(f" Action: {interaction.get('recommended_action')}")
else:
print("✓ No significant interactions found")
# ============================================================================
# EXAMPLE 4: FHIR Resource Creation
# ============================================================================
def example_fhir_resources():
"""Example: Create and work with FHIR resources"""
print("\n" + "="*70)
print("EXAMPLE 4: FHIR Resource Creation")
print("="*70)
# Create patient resource
print("\nCreating FHIR Patient resource...")
patient = FHIRPatient(
id="pat-001",
name="John Doe",
birth_date="1980-05-15",
gender="male",
mrn="MRN001",
contact_number="555-1234"
)
patient_fhir = patient.to_fhir()
print(f"✓ Patient age: {patient.get_age()} years")
print(f"✓ FHIR ID: {patient_fhir['id']}")
print(f"✓ MRN: {patient_fhir['identifier'][0]['value']}")
# Create observation resource
print("\nCreating FHIR Observation (Lab) resource...")
lab = FHIRObservation(
id="obs-001",
code="2345-7",
code_display="Glucose",
value=145,
unit="mg/dL",
reference_low=70,
reference_high=100,
effective_datetime="2024-01-15T09:00:00Z"
)
lab_fhir = lab.to_fhir()
print(f"✓ Test: {lab.code_display}")
print(f"✓ Value: {lab.value} {lab.unit}")
print(f"✓ Status: {'ABNORMAL' if lab.is_abnormal() else 'NORMAL'}")
print(f"✓ Interpretation: {lab.get_interpretation()}")
# ============================================================================
# EXAMPLE 5: Clinical Scoring
# ============================================================================
def example_clinical_scores():
"""Example: Calculate clinical risk scores"""
print("\n" + "="*70)
print("EXAMPLE 5: Clinical Risk Scores")
print("="*70)
# QSOFA Score (Sepsis Risk)
print("\nCalculating QSOFA Score (Sepsis Risk)...")
qsofa = ClinicalScores.qsofa_score(
altered_mental_status=False,
systolic_bp=98, # Low
respiratory_rate=24 # High
)
risk_level = "HIGH" if qsofa >= 2 else "LOW"
print(f"✓ QSOFA Score: {qsofa}/3")
print(f"✓ Sepsis Risk: {risk_level}")
print(f" ⚠️ Score ≥2 suggests high risk of sepsis; consider screening")
# APACHE II Score (Mortality Risk)
print("\nCalculating APACHE II Score (ICU Mortality Risk)...")
apache_score, mortality_risk = ClinicalScores.apache_ii_score(
age=72,
heart_rate=95,
mean_arterial_pressure=85,
temperature_celsius=37.5,
respiratory_rate=18,
ph=7.40,
hematocrit=38,
wbc=8.5,
glasgow_coma_score=15
)
print(f"✓ APACHE II Score: {apache_score}")
print(f"✓ Estimated Mortality Risk: {mortality_risk}")
# ============================================================================
# EXAMPLE 6: PII Redaction
# ============================================================================
def example_pii_redaction():
"""Example: Redact PII from text"""
print("\n" + "="*70)
print("EXAMPLE 6: PII Redaction for Audit Logs")
print("="*70)
original_text = """
Patient John Doe (SSN 123-45-6789) with MRN MRN001 was admitted.
Contact: john.doe@example.com or 555-123-4567
Credit card ending in 4532-1111-2222-3333 on file.
Date of birth: 1980-05-15
"""
print("\nOriginal text:")
print(original_text)
redacted = PIIRedactor.redact_text(original_text)
print("\nRedacted text:")
print(redacted)
# ============================================================================
# EXAMPLE 7: Medication Interactions Database
# ============================================================================
def example_medication_database():
"""Example: Query medication database"""
print("\n" + "="*70)
print("EXAMPLE 7: Medication Database Query")
print("="*70)
# Get drug info
print("\nLooking up Metformin...")
metformin = MedicationDatabase.get_drug_info("Metformin")
if metformin:
print(f"✓ Drug: Metformin")
print(f" Generic: {metformin['generic_name']}")
print(f" Class: {metformin['classification']}")
print(f" Contraindications: {', '.join(metformin['contraindications'])}")
print(f" Known Interactions: {len(metformin['interactions'])}")
# Check specific interaction
print("\nChecking Metformin + NSAIDs interaction...")
interaction = MedicationDatabase.check_interaction("Metformin", "NSAIDs")
if interaction:
print(f"⚠️ Interaction found!")
print(f" Severity: {interaction['severity'].upper()}")
print(f" Description: {interaction['description']}")
# ============================================================================
# EXAMPLE 8: Authorization Scenario
# ============================================================================
async def example_authorization():
"""Example: Different user roles and permissions"""
print("\n" + "="*70)
print("EXAMPLE 8: Role-Based Access Control (RBAC)")
print("="*70)
# Test different roles
roles = ["physician", "nurse", "pharmacist", "patient"]
tools = ["get_patient_vitals", "order_medication", "check_interactions"]
print("\nAccess Control Matrix:")
print(f"{'Role':<15} {'Get Vitals':<15} {'Order Med':<15} {'Check Interact':<15}")
print("-" * 60)
from mcp_healthcare_server import AuthorizationManager
for role in roles:
row = f"{role:<15}"
for tool in tools:
can_access = AuthorizationManager.can_access(role, tool)
status = "✓ Yes" if can_access else "✗ No"
row += f"{status:<15}"
print(row)
# ============================================================================
# EXAMPLE 9: Complete Clinical Scenario
# ============================================================================
async def example_clinical_scenario():
"""Example: Complete patient encounter scenario"""
print("\n" + "="*70)
print("EXAMPLE 9: Complete Clinical Scenario")
print("="*70)
context = {
"user_id": "doc-001",
"user_role": "physician"
}
print("\nScenario: Evaluating patient MRN001 presenting with possible infection")
# Step 1: Get vitals
print("\n[Step 1] Get vital signs...")
vitals_result = await get_patient_vitals({"patient_id": "MRN001"}, context)
print(f"✓ Vitals retrieved")
# Step 2: Get labs
print("\n[Step 2] Get lab results...")
labs_result = await get_patient_labs({"patient_id": "MRN001"}, context)
print(f"✓ Labs retrieved")
# Step 3: Get medications
print("\n[Step 3] Get current medications...")
meds_result = await get_patient_medications({"patient_id": "MRN001"}, context)
meds = meds_result.get("medications", [])
print(f"✓ Patient on {len(meds)} medications")
# Step 4: Check interactions before ordering new med
print("\n[Step 4] Check drug interactions...")
drug_names = [m["drug_name"] for m in meds]
drug_names.append("Azithromycin") # Proposed antibiotic
interaction_result = await check_drug_interactions({"drugs": drug_names}, context)
interactions = interaction_result.get("interactions", [])
print(f"⚠️ Found {len(interactions)} potential interactions")
# Step 5: Clinical assessment
print("\n[Step 5] Clinical Assessment:")
patient = await ehr_client.get_patient("MRN001")
if patient:
age = datetime.now().year - int(patient.birth_date[:4])
print(f" - Patient age: {age} years")
# Step 6: Documentation
print("\n[Step 6] Generate clinical note...")
note_result = await generate_discharge_summary({"patient_id": "MRN001"}, context)
if note_result.get("success"):
print("✓ Clinical note generated")
print("\n" + "="*70)
print("Scenario complete. Ready for clinical decision-making.")
print("="*70)
# ============================================================================
# MAIN TEST SUITE
# ============================================================================
async def run_all_examples():
"""Run all examples"""
print("\n" + "🏥" * 35)
print("MCP HEALTHCARE SYSTEM - COMPREHENSIVE EXAMPLES")
print("🏥" * 35)
# Synchronous examples
example_fhir_resources()
example_clinical_scores()
example_pii_redaction()
example_medication_database()
example_authorization()
# Async examples (if services available)
try:
await example_patient_retrieval()
await example_lab_interpretation()
await example_medication_management()
await example_clinical_scenario()
except Exception as e:
print(f"\n⚠️ Async examples skipped (services may not be running): {e}")
print("\n" + "="*70)
print("✓ All examples completed!")
print("="*70)
# ============================================================================
# UNIT TESTS
# ============================================================================
def test_fhir_observation():
"""Test FHIR Observation resource"""
obs = FHIRObservation(
id="test-1",
code="2345-7",
code_display="Glucose",
value=85,
unit="mg/dL",
reference_low=70,
reference_high=100,
effective_datetime="2024-01-15T09:00:00Z"
)
assert obs.get_interpretation() == "N", "Normal value should return 'N'"
assert not obs.is_abnormal(), "Normal value should not be abnormal"
# Test abnormal high
obs.value = 150
assert obs.get_interpretation() == "H", "High value should return 'H'"
assert obs.is_abnormal(), "High value should be abnormal"
print("✓ test_fhir_observation passed")
def test_pii_redaction():
"""Test PII redaction"""
text = "SSN 123-45-6789"
redacted = PIIRedactor.redact_text(text, ["ssn"])
assert "123-45-6789" not in redacted, "SSN should be redacted"
assert "XXX-XX-6789" in redacted, "SSN should be masked correctly"
print("✓ test_pii_redaction passed")
def test_clinical_scores():
"""Test clinical score calculations"""
# Test QSOFA
qsofa = ClinicalScores.qsofa_score(
altered_mental_status=True,
systolic_bp=95,
respiratory_rate=25
)
assert qsofa == 3, "QSOFA should be 3 for all positive findings"
print("✓ test_clinical_scores passed")
def test_medication_interactions():
"""Test medication interaction lookup"""
interaction = MedicationDatabase.check_interaction("Warfarin", "NSAIDs")
assert interaction is not None, "Warfarin-NSAID interaction should exist"
assert interaction["severity"] == "high", "Warfarin-NSAID should be high severity"
print("✓ test_medication_interactions passed")
# ============================================================================
# MAIN
# ============================================================================
if __name__ == "__main__":
import sys
if len(sys.argv) > 1 and sys.argv[1] == "test":
print("\n" + "="*70)
print("Running Unit Tests")
print("="*70 + "\n")
test_fhir_observation()
test_pii_redaction()
test_clinical_scores()
test_medication_interactions()
print("\n✓ All tests passed!")
else:
# Run examples
asyncio.run(run_all_examples())
