🏥 MCP Healthcare System - Production-Grade Python Implementation
Complete Model Context Protocol (MCP) server for healthcare AI systems with Claude integration, FHIR support, HIPAA compliance, and production-ready deployment patterns.
📋 Table of Contents
- Quick Start
- Architecture
- Project Structure
- Installation
- Configuration
- Usage Examples
- Clinical Tools
- Security & Compliance
- Deployment
- Testing
- API Documentation
🚀 Quick Start
Prerequisites
- Python 3.11+
- Docker & Docker Compose (optional, for containerized deployment)
- ANTHROPIC_API_KEY environment variable set
Installation
# Clone the repository
git clone <repo-url>
cd mcp-healthcare
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install dependencies
pip install -r requirements.txt
# Set environment variables
export ANTHROPIC_API_KEY="your-key-here"
export ENVIRONMENT="development"
Run Examples
# Run all examples
python examples.py
# Run unit tests
python examples.py test
# Run MCP server
python mcp_healthcare_server.py
# Run client (queries Claude)
python mcp_healthcare_client.py
Using Docker Compose
# Start all services (MCP, PostgreSQL, Redis, Jaeger, Prometheus, Grafana)
docker-compose up -d
# View logs
docker-compose logs -f mcp-server
# Access services
# - MCP Server: http://localhost:3000
# - Jaeger UI: http://localhost:16686
# - Prometheus: http://localhost:9090
# - Grafana: http://localhost:3001 (admin/admin)
# Stop services
docker-compose down
🏗️ Architecture
┌─────────────────────────────────────────────────────────────────┐
│ Claude (LLM) │
└────────────────────┬────────────────────────────────────────────┘
│ (Tool Calls)
▼
┌─────────────────────────────────────────────────────────────────┐
│ MCP Protocol Handler │
│ - Tool Registry & Execution │
│ - Authorization (RBAC) │
│ - Audit Logging │
└────────────┬──────────────┬──────────────┬──────────────────────┘
│ │ │
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ EHR APIs │ │ Databases │ │ Resources │
│ - Vitals │ │ - Audit │ │ - Guidelines│
│ - Labs │ │ - Cache │ │ - Templates │
│ - Meds │ │ - Sessions │ │ - Drug DB │
└──────────────┘ └──────────────┘ └──────────────┘
📁 Project Structure
mcp-healthcare/
├── mcp_healthcare_server.py # Core MCP server implementation
├── mcp_healthcare_client.py # Claude integration client
├── config.py # Configuration management
├── healthcare_utils.py # FHIR, clinical scoring, utilities
├── examples.py # Comprehensive examples & tests
├── requirements.txt # Python dependencies
├── Dockerfile # Container image
├── docker-compose.yml # Local development stack
├── README.md # This file
├── init-db.sql # Database initialization
├── prometheus.yml # Prometheus config
├── grafana-datasources.yml # Grafana config
└── tests/
├── test_server.py # Server unit tests
├── test_tools.py # Tool execution tests
├── test_security.py # Security/RBAC tests
└── test_integration.py # End-to-end integration tests
⚙️ Configuration
Environment Variables
# Application
ENVIRONMENT=development # local, development, staging, production
ANTHROPIC_API_KEY=sk-... # Your Claude API key
# EHR Backend
EHR_API_URL=http://localhost:8080
EHR_API_KEY=your-ehr-key
# Database
DB_HOST=localhost
DB_PORT=5432
DB_NAME=clinical_mcp
DB_USER=postgres
DB_PASSWORD=postgres
# Redis
REDIS_URL=redis://localhost:6379
# AWS (optional)
AWS_REGION=us-east-1
AWS_ACCESS_KEY_ID=***
AWS_SECRET_ACCESS_KEY=***
# Security
ENCRYPTION_KEY_ID=arn:aws:kms:...
Configuration Files
See config.py for comprehensive configuration options:
from config import get_config
config = get_config()
print(config.server.ehr_api_timeout) # 5.0 seconds
print(config.security.enable_hipaa_audit) # True
print(config.deployment.k8s_namespace) # "healthcare"
💊 Clinical Tools
Available Tools
1. get_patient_vitals
Retrieve vital signs (HR, BP, Temp, O2, RR)
result = await get_patient_vitals({
"patient_id": "MRN001",
"limit": 10
}, context)
2. get_patient_labs
Retrieve laboratory results with FHIR Observation resources
result = await get_patient_labs({
"patient_id": "MRN001",
"limit": 20
}, context)
3. get_patient_medications
Get active medications with dose, frequency, status
result = await get_patient_medications({
"patient_id": "MRN001",
"include_discontinued": False
}, context)
4. check_drug_interactions
Identify drug-drug interactions with severity levels
result = await check_drug_interactions({
"drugs": ["Metformin", "NSAIDs", "Lisinopril"]
}, context)
5. order_medication
Place medication order with safety checks (includes sampling)
result = await order_medication({
"patient_id": "MRN001",
"drug_name": "Metformin",
"dose": 500,
"unit": "mg"
}, context)
6. generate_discharge_summary
Generate clinical documentation from patient data
result = await generate_discharge_summary({
"patient_id": "MRN001"
}, context)
🔐 Security & Compliance
HIPAA Compliance
✅ Audit Logging
- Immutable audit trail of all tool calls
- 7-year retention (configurable)
- Masked PII in logs
audit_logger.log_tool_execution(
user_id="doc-001",
user_role="physician",
tool_name="get_patient_labs",
patient_id="MRN001",
resource_type="labs",
authorized=True,
result="success"
)
✅ PII Redaction
- Automatic masking of SSN, phone, email, MRN, credit cards
- Configurable redaction patterns
from healthcare_utils import PIIRedactor
redacted = PIIRedactor.redact_text(
"Patient SSN 123-45-6789",
patterns=["ssn"]
)
# Result: "Patient SSN XXX-XX-6789"
✅ Role-Based Access Control (RBAC)
- Fine-grained permissions by role (physician, nurse, pharmacist, patient)
- Tool-level access control
from mcp_healthcare_server import AuthorizationManager
can_order = AuthorizationManager.can_access("pharmacist", "order_medication")
# True
✅ Encryption
- TLS 1.3 for all network communications
- AES-256 encryption at rest
- KMS key management
Data Protection
- ✅ Input validation on all parameters
- ✅ PII filtering before response to LLM
- ✅ Rate limiting (per-user, per-tool)
- ✅ Timeout protection (prevent hanging requests)
- ✅ Circuit breaker for external APIs
🚢 Deployment
Docker Deployment
Build and run with Docker:
# Build image
docker build -t clinical-mcp:latest .
# Run container
docker run -p 3000:3000 \
-e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY \
-e DB_HOST=postgres \
clinical-mcp:latest
Kubernetes Deployment
# Create namespace
kubectl create namespace healthcare
# Apply configuration
kubectl apply -f k8s/deployment.yaml -n healthcare
# Scale replicas
kubectl scale deployment clinical-mcp --replicas=5 -n healthcare
# View logs
kubectl logs -f deployment/clinical-mcp -n healthcare
See k8s/ directory for complete Kubernetes manifests.
Production Checklist
- Set
ENVIRONMENT=productionin config - Enable HTTPS/TLS
- Configure database with backups
- Set up distributed tracing (Jaeger/Datadog)
- Enable Prometheus metrics
- Configure Grafana dashboards
- Set up alerting rules
- Enable audit logging to S3
- Configure auto-scaling (HPA)
- Set up multi-region deployment
- Run security scan:
bandit -r . - Run dependency check:
safety check
🧪 Testing
Run Examples
# Run comprehensive examples
python examples.py
# Run specific example
python -c "from examples import example_patient_retrieval; asyncio.run(example_patient_retrieval())"
Run Unit Tests
# Run all tests
pytest tests/ -v
# Run with coverage
pytest tests/ --cov=src --cov-report=html
# Run specific test
pytest tests/test_security.py::test_rbac -v
Integration Tests
# Start services
docker-compose up -d
# Run integration tests
pytest tests/test_integration.py -v --integration
# Stop services
docker-compose down
📊 Monitoring & Observability
Prometheus Metrics
Available at http://localhost:9090:
clinical_mcp_tool_calls_total # Total tool calls
clinical_mcp_tool_execution_seconds # Tool execution latency
clinical_mcp_authorization_failures_total # Auth failures
clinical_mcp_audit_logs_total # Audit log entries
Grafana Dashboards
Available at http://localhost:3001 (admin/admin):
- System Health: CPU, memory, request rates
- Tool Performance: Latency, error rates, throughput
- Security: Authorization attempts, audit logs
- Clinical Metrics: Patient encounters, labs processed
Jaeger Tracing
Available at http://localhost:16686:
Trace complete request flows through the system with:
- Tool execution spans
- Database query timing
- External API calls
- Error propagation
📚 Clinical Examples
Scenario 1: Patient Admission Vitals Review
# Clinician reviews patient vitals on admission
query = """Get the vital signs for patient MRN001.
Are there any concerning values?
Flag any abnormal findings."""
response = await processor.process_query(query)
Expected flow:
- Claude calls
get_patient_vitals - Server retrieves FHIR Observation resources
- Claude interprets findings
- Returns clinical assessment
Scenario 2: Medication Reconciliation
query = """Patient MRN001 is admitted with new medications.
Get their current home medications and check for interactions
with: Azithromycin (500mg daily), Dexamethasone (4mg QID)"""
response = await processor.process_query(query)
Expected flow:
- Claude calls
get_patient_medications - Claude calls
check_drug_interactions - Claude identifies significant interactions
- Returns safety assessment and recommendations
Scenario 3: Clinical Documentation
query = """Generate discharge summary for patient MRN001
and include current medication list and follow-up recommendations."""
response = await processor.process_query(query)
Expected flow:
- Claude calls
get_patient_medications - Claude calls
generate_discharge_summary - Returns formatted clinical document
- Ready for physician review/signature
🔗 Integration Points
External Systems
MCP Server → EHR API (Epic, Cerner, Athena)
→ Cloud Storage (S3, GCS)
→ Database (PostgreSQL, MySQL)
→ Cache (Redis)
→ Secret Manager (AWS Secrets Manager)
→ Monitoring (Datadog, New Relic)
Common Integrations
EHR Systems:
- Epic via HL7 FHIR APIs
- Cerner via SMART on FHIR
- Athena via Anthem APIs
Cloud Providers:
- AWS: Lambda, RDS, S3, KMS, CloudWatch
- GCP: Cloud Functions, Cloud SQL, Cloud Storage
- Azure: App Service, SQL Database, Key Vault
Monitoring:
- Datadog
- New Relic
- Splunk
- CloudWatch
🐛 Troubleshooting
Common Issues
"ANTHROPIC_API_KEY not set"
export ANTHROPIC_API_KEY="sk-..."
"Connection refused" on EHR API
# Check if mock EHR is running
curl http://localhost:8080/health
# Or use docker-compose
docker-compose up -d ehr-mock
"Database connection error"
# Check PostgreSQL is running
docker-compose ps postgres
# Check connection string
echo $DATABASE_URL
High latency on tool calls
# Reduce timeout threshold
config.server.ehr_api_timeout = 2.0
# Check network latency
ping ehr-api-host
📖 API Documentation
MCP Protocol
Full OpenRPC specification at:
http://localhost:3000/rpc/openrpc.json
Claude Integration
See mcp_healthcare_client.py for complete implementation:
from mcp_healthcare_client import ClinicalQueryProcessor
processor = ClinicalQueryProcessor(api_key="sk-...")
result = await processor.process_query(
"What are patient MRN001's current medications?"
)
📝 FHIR Resources
Supported FHIR R4 resources:
- Patient - Demographics
- Observation - Labs, vitals, findings
- MedicationRequest - Prescriptions
- Medication - Drug information
- Condition - Problems/diagnoses
- AllergyIntolerance - Allergies
- Encounter - Visits/admissions
- Procedure - Surgical/medical procedures
- DiagnosticReport - Test reports
🤝 Contributing
# Create feature branch
git checkout -b feature/new-tool
# Make changes and test
pytest tests/ -v
# Submit PR
git push origin feature/new-tool
📄 License
This project is licensed under the MIT License - see LICENSE file for details.
🆘 Support
- Documentation: See
/docsdirectory - Issues: GitHub Issues
- Email: support@example.com
- Slack: #healthcare-ai channel
🎓 Learning Resources
Built with ❤️ for Healthcare AI
For production deployment, contact the enterprise team.