# 🏥 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](#quick-start)
- [Architecture](#architecture)
- [Project Structure](#project-structure)
- [Installation](#installation)
- [Configuration](#configuration)
- [Usage Examples](#usage-examples)
- [Clinical Tools](#clinical-tools)
- [Security & Compliance](#security--compliance)
- [Deployment](#deployment)
- [Testing](#testing)
- [API Documentation](#api-documentation)
## 🚀 Quick Start
### Prerequisites
- Python 3.11+
- Docker & Docker Compose (optional, for containerized deployment)
- ANTHROPIC_API_KEY environment variable set
### Installation
```bash
# 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
```bash
# 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
```bash
# 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
```bash
# 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:
```python
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)
```python
result = await get_patient_vitals({
"patient_id": "MRN001",
"limit": 10
}, context)
```
#### 2. **get_patient_labs**
Retrieve laboratory results with FHIR Observation resources
```python
result = await get_patient_labs({
"patient_id": "MRN001",
"limit": 20
}, context)
```
#### 3. **get_patient_medications**
Get active medications with dose, frequency, status
```python
result = await get_patient_medications({
"patient_id": "MRN001",
"include_discontinued": False
}, context)
```
#### 4. **check_drug_interactions**
Identify drug-drug interactions with severity levels
```python
result = await check_drug_interactions({
"drugs": ["Metformin", "NSAIDs", "Lisinopril"]
}, context)
```
#### 5. **order_medication**
Place medication order with safety checks (includes sampling)
```python
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
```python
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
```python
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
```python
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
```python
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:
```bash
# 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
```bash
# 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=production` in 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
```bash
# 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
```bash
# 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
```bash
# 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
```python
# 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:
1. Claude calls `get_patient_vitals`
2. Server retrieves FHIR Observation resources
3. Claude interprets findings
4. Returns clinical assessment
### Scenario 2: Medication Reconciliation
```python
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:
1. Claude calls `get_patient_medications`
2. Claude calls `check_drug_interactions`
3. Claude identifies significant interactions
4. Returns safety assessment and recommendations
### Scenario 3: Clinical Documentation
```python
query = """Generate discharge summary for patient MRN001
and include current medication list and follow-up recommendations."""
response = await processor.process_query(query)
```
Expected flow:
1. Claude calls `get_patient_medications`
2. Claude calls `generate_discharge_summary`
3. Returns formatted clinical document
4. 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"**
```bash
export ANTHROPIC_API_KEY="sk-..."
```
**"Connection refused" on EHR API**
```bash
# Check if mock EHR is running
curl http://localhost:8080/health
# Or use docker-compose
docker-compose up -d ehr-mock
```
**"Database connection error"**
```bash
# Check PostgreSQL is running
docker-compose ps postgres
# Check connection string
echo $DATABASE_URL
```
**High latency on tool calls**
```python
# 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:
```python
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
```bash
# 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 `/docs` directory
- **Issues**: GitHub Issues
- **Email**: support@example.com
- **Slack**: #healthcare-ai channel
## 🎓 Learning Resources
- [FHIR Documentation](https://www.hl7.org/fhir/)
- [HL7 Standards](https://www.hl7.org/)
- [Claude API Docs](https://docs.anthropic.com/)
- [MCP Protocol Spec](https://modelcontextprotocol.io/)
- [HIPAA Compliance Guide](https://www.hhs.gov/hipaa/)
---
**Built with ❤️ for Healthcare AI**
For production deployment, contact the enterprise team.
