Epic Patient API MCP Server

"""Patient data loader for YAML files.""" import yaml from pathlib import Path from typing import List, Optional from ..types import ( Demographics, Allergy, Medication, ClinicalNote, LabResult, VitalSigns, Condition, Procedure, ) class PatientDataLoader: """Loads patient data from YAML files.""" def __init__(self, patients_dir: str = "patients"): self.patients_dir = Path(patients_dir) def _load_yaml(self, patient_id: str, filename: str) -> Optional[List]: """Load a YAML file for a patient.""" file_path = self.patients_dir / patient_id / f"{filename}.yaml" if not file_path.exists(): return None with open(file_path, 'r') as f: data = yaml.safe_load(f) return data if data else [] def get_demographics(self, patient_id: str) -> Optional[Demographics]: """Load patient demographics.""" data = self._load_yaml(patient_id, "demographics") if not data: return None return Demographics(**data) def get_allergies(self, patient_id: str) -> List[Allergy]: """Load patient allergies.""" data = self._load_yaml(patient_id, "allergies") if not data: return [] return [Allergy(**item) for item in data] def get_medications(self, patient_id: str, status: Optional[str] = None) -> List[Medication]: """Load patient medications, optionally filtered by status.""" data = self._load_yaml(patient_id, "medications") if not data: return [] medications = [Medication(**item) for item in data] if status: medications = [med for med in medications if med.status == status] return medications def get_clinical_notes( self, patient_id: str, note_type: Optional[str] = None ) -> List[ClinicalNote]: """Load clinical notes, optionally filtered by type.""" data = self._load_yaml(patient_id, "clinical_notes") if not data: return [] notes = [ClinicalNote(**item) for item in data] if note_type: notes = [note for note in notes if note.type == note_type] return notes def get_attachment(self, patient_id: str, attachment_id: str) -> Optional[str]: """Load attachment content by ID.""" attachment_path = self.patients_dir / patient_id / "attachments" / f"{attachment_id}.txt" if not attachment_path.exists(): return None with open(attachment_path, 'r') as f: return f.read() def get_lab_results(self, patient_id: str) -> List[LabResult]: """Load lab results.""" data = self._load_yaml(patient_id, "labs") if not data: return [] return [LabResult(**item) for item in data] def get_vitals(self, patient_id: str) -> List[VitalSigns]: """Load vital signs.""" data = self._load_yaml(patient_id, "vitals") if not data: return [] return [VitalSigns(**item) for item in data] def get_conditions(self, patient_id: str) -> List[Condition]: """Load patient conditions/diagnoses.""" data = self._load_yaml(patient_id, "conditions") if not data: return [] return [Condition(**item) for item in data] def get_procedures(self, patient_id: str) -> List[Procedure]: """Load procedures.""" data = self._load_yaml(patient_id, "procedures") if not data: return [] return [Procedure(**item) for item in data] def get_all_patient_data(self, patient_id: str) -> dict: """Load all data for a patient as a dictionary.""" return { "demographics": self.get_demographics(patient_id), "allergies": self.get_allergies(patient_id), "medications": self.get_medications(patient_id), "conditions": self.get_conditions(patient_id), "clinical_notes": self.get_clinical_notes(patient_id), "lab_results": self.get_lab_results(patient_id), "vitals": self.get_vitals(patient_id), "procedures": self.get_procedures(patient_id), } def patient_exists(self, patient_id: str) -> bool: """Check if a patient directory exists.""" return (self.patients_dir / patient_id).exists()