<div align="center" style="margin-bottom: 1em;">
# HealthChain 💫 🏥
<img src="https://raw.githubusercontent.com/dotimplement/HealthChain/main/docs/assets/images/healthchain_logo.png" alt="HealthChain Logo" width=300></img>
 
</div>
Connect your AI models to any healthcare system with a few lines of Python 💫 🏥.
Integrating AI with electronic health records (EHRs) is complex, manual, and time-consuming. Let's try to change that.
```bash
pip install healthchain
```
First time here? Check out our [Docs](https://dotimplement.github.io/HealthChain/) page!
## Features
- [x] 🔌 **Gateway**: Connect to multiple EHR systems with [unified API](https://dotimplement.github.io/HealthChain/reference/gateway/gateway/) supporting FHIR, CDS Hooks, and SOAP/CDA protocols (sync / async support)
- [x] 🔥 **Pipelines**: Build FHIR-native ML workflows or use [pre-built ones](https://dotimplement.github.io/HealthChain/reference/pipeline/pipeline/#prebuilt) for your healthcare NLP and AI tasks
- [x] 🔄 **InteropEngine**: Convert between FHIR, CDA, and HL7v2 with a [template-based engine](https://dotimplement.github.io/HealthChain/reference/interop/interop/)
- [x] 🔒 Type-safe healthcare data with full type hints and Pydantic validation for [FHIR resources](https://dotimplement.github.io/HealthChain/reference/utilities/fhir_helpers/)
- [x] ⚡ Built-in event-driven logging and operation tracking for [audit trails](https://dotimplement.github.io/HealthChain/reference/gateway/events/)
- [x] 🚀 Deploy production-ready applications with [HealthChainAPI](https://dotimplement.github.io/HealthChain/reference/gateway/api/) and FastAPI integration
- [x] 🧪 Generate [synthetic healthcare data](https://dotimplement.github.io/HealthChain/reference/utilities/data_generator/) and [sandbox testing](https://dotimplement.github.io/HealthChain/reference/sandbox/sandbox/) utilities
- [x] 🖥️ Bootstrap configurations with CLI tools
## Why use HealthChain?
- **EHR integrations are manual and time-consuming** - **HealthChainAPI** abstracts away complexities so you can focus on AI development, not learning FHIR APIs, CDS Hooks, and authentication schemes.
- **Healthcare data is fragmented and complex** - **InteropEngine** handles the conversion between FHIR, CDA, and HL7v2 so you don't have to become an expert in healthcare data standards.
- [**Most healthcare data is unstructured**](https://pmc.ncbi.nlm.nih.gov/articles/PMC10566734/) - HealthChain **Pipelines** are optimized for real-time AI and NLP applications that deal with realistic healthcare data.
- **Built by health tech developers, for health tech developers** - HealthChain is tech stack agnostic, modular, and easily extensible with built-in compliance and audit features.
## HealthChainAPI
The HealthChainAPI provides a secure integration layer that coordinates multiple healthcare systems in a single application.
### Multi-Protocol Support
Connect to multiple healthcare data sources and protocols:
```python
from healthchain.gateway import (
HealthChainAPI, FHIRGateway,
CDSHooksService, NoteReaderService
)
# Create your healthcare application
app = HealthChainAPI(
title="My Healthcare AI App",
description="AI-powered patient care platform"
)
# FHIR for patient data from multiple EHRs
fhir = FHIRGateway()
fhir.add_source("epic", "fhir://fhir.epic.com/r4?client_id=...")
fhir.add_source("medplum", "fhir://api.medplum.com/fhir/R4/?client_id=...")
# CDS Hooks for real-time clinical decision support
cds = CDSHooksService()
@cds.hook("patient-view", id="allergy-alerts")
def check_allergies(request):
# Your AI logic here
return {"cards": [...]}
# SOAP for clinical document processing
notes = NoteReaderService()
@notes.method("ProcessDocument")
def process_note(request):
# Your NLP pipeline here
return processed_document
# Register everything
app.register_gateway(fhir)
app.register_service(cds)
app.register_service(notes)
# Your API now handles:
# /fhir/* - Patient data, observations, etc.
# /cds/* - Real-time clinical alerts
# /soap/* - Clinical document processing
# Deploy with uvicorn
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, port=8888)
```
### FHIR Operations with AI Enhancement
```python
from healthchain.gateway import FHIRGateway
from fhir.resources.patient import Patient
gateway = FHIRGateway()
gateway.add_source("epic", "fhir://fhir.epic.com/r4?...")
# Add AI transformations to FHIR data
@gateway.transform(Patient)
def enhance_patient(id: str, source: str = None) -> Patient:
patient = gateway.read(Patient, id, source)
# Get lab results and process with AI
lab_results = gateway.search(
Observation,
{"patient": id, "category": "laboratory"},
source
)
insights = nlp_pipeline.process(patient, lab_results)
# Add AI summary to patient record
patient.extension = patient.extension or []
patient.extension.append({
"url": "http://healthchain.org/fhir/summary",
"valueString": insights.summary
})
# Update the patient record
gateway.update(patient, source)
return patient
# Automatically available at: GET /fhir/transform/Patient/123?source=epic
```
## Pipeline
Pipelines provide a flexible way to build and manage processing pipelines for NLP and ML tasks that can easily integrate with complex healthcare systems.
### Building a pipeline
```python
from healthchain.io.containers import Document
from healthchain.pipeline import Pipeline
from healthchain.pipeline.components import (
TextPreProcessor,
SpacyNLP,
TextPostProcessor,
)
# Initialize the pipeline
nlp_pipeline = Pipeline[Document]()
# Add TextPreProcessor component
preprocessor = TextPreProcessor()
nlp_pipeline.add_node(preprocessor)
# Add Model component (assuming we have a pre-trained model)
spacy_nlp = SpacyNLP.from_model_id("en_core_sci_sm")
nlp_pipeline.add_node(spacy_nlp)
# Add TextPostProcessor component
postprocessor = TextPostProcessor(
postcoordination_lookup={
"heart attack": "myocardial infarction",
"high blood pressure": "hypertension",
}
)
nlp_pipeline.add_node(postprocessor)
# Build the pipeline
nlp = nlp_pipeline.build()
# Use the pipeline
result = nlp(Document("Patient has a history of heart attack and high blood pressure."))
print(f"Entities: {result.nlp.get_entities()}")
```
#### Working with healthcare data formats
Adapters handle conversion between healthcare formats (CDA, FHIR) and internal Document objects for seamless EHR integration.
```python
from healthchain.io import CdaAdapter, Document
from healthchain.models import CdaRequest
adapter = CdaAdapter()
# Parse healthcare data into Document
cda_request = CdaRequest(document="<CDA XML content>")
doc = adapter.parse(cda_request)
# Process with your pipeline
processed_doc = nlp_pipeline(doc)
# Access extracted clinical data
print(f"Problems: {processed_doc.fhir.problem_list}")
print(f"Medications: {processed_doc.fhir.medication_list}")
# Convert back to healthcare format
response = adapter.format(processed_doc)
```
### Using pre-built pipelines
Pre-built pipelines are use case specific end-to-end workflows optimized for common healthcare AI tasks.
```python
from healthchain.pipeline import MedicalCodingPipeline
from healthchain.models import CdaRequest
# Load from model ID
pipeline = MedicalCodingPipeline.from_model_id(
model="blaze999/Medical-NER", task="token-classification", source="huggingface"
)
# Simple end-to-end processing
cda_request = CdaRequest(document="<CDA XML content>")
response = pipeline.process_request(cda_request)
# Or manual control for document access
from healthchain.io import CdaAdapter
adapter = CdaAdapter()
doc = adapter.parse(cda_request)
doc = pipeline(doc)
# Access: doc.fhir.problem_list, doc.fhir.medication_list
response = adapter.format(doc)
```
## Interoperability
The InteropEngine is a template-based system that allows you to convert between FHIR, CDA, and HL7v2.
```python
from healthchain.interop import create_interop, FormatType
engine = create_interop()
with open("tests/data/test_cda.xml", "r") as f:
cda_data = f.read()
# Convert CDA to FHIR
fhir_resources = engine.to_fhir(cda_data, src_format=FormatType.CDA)
# Convert FHIR to CDA
cda_data = engine.from_fhir(fhir_resources, dest_format=FormatType.CDA)
```
## Sandbox
Test your AI applications in realistic healthcare contexts with [CDS Hooks](https://cds-hooks.org/) sandbox environments.
```python
import healthchain as hc
from healthchain.sandbox.use_cases import ClinicalDecisionSupport
@hc.sandbox
class MyCDS(ClinicalDecisionSupport):
def __init__(self):
self.pipeline = SummarizationPipeline.from_model_id("facebook/bart-large-cnn")
@hc.ehr(workflow="encounter-discharge")
def ehr_database_client(self):
return self.data_generator.generate_prefetch()
cds = MyCDS()
cds.start_sandbox()
# Run with: healthchain run mycds.py
```
## Road Map
- [ ] 🔒 Built-in HIPAA compliance validation and PHI detection
- [ ] 📊 Track configurations, data provenance, and monitor model performance with MLFlow integration
- [ ] 🚀 Compliance monitoring, auditing at deployment as a sidecar service
- [ ] 🔄 HL7v2 parsing and FHIR profile conversion support
- [ ] 🧠 Multi-modal pipelines
## Contribute
We are always eager to hear feedback and suggestions, especially if you are a developer or researcher working with healthcare systems!
- 💡 Let's chat! [Discord](https://discord.gg/UQC6uAepUz)
- 🛠️ [Contribution Guidelines](CONTRIBUTING.md)
## Acknowledgements 🤗
This project builds on [fhir.resources](https://github.com/nazrulworld/fhir.resources) and [CDS Hooks](https://cds-hooks.org/) standards developed by [HL7](https://www.hl7.org/) and [Boston Children's Hospital](https://www.childrenshospital.org/).
Raw data
{
"_id": null,
"home_page": null,
"name": "healthchain",
"maintainer": null,
"docs_url": null,
"requires_python": "<3.12,>=3.9",
"maintainer_email": null,
"keywords": "nlp, ai, llm, healthcare, ehr, mlops",
"author": "Jennifer Jiang-Kells",
"author_email": "jenniferjiangkells@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/6f/7d/6429cbe147a88bbbc25ec5fe823c5175c5959c1f60682d9a28cf58970648/healthchain-0.10.0.tar.gz",
"platform": null,
"description": "<div align=\"center\" style=\"margin-bottom: 1em;\">\n\n# HealthChain \ud83d\udcab \ud83c\udfe5\n\n<img src=\"https://raw.githubusercontent.com/dotimplement/HealthChain/main/docs/assets/images/healthchain_logo.png\" alt=\"HealthChain Logo\" width=300></img>\n\n\n \n\n\n</div>\n\nConnect your AI models to any healthcare system with a few lines of Python \ud83d\udcab \ud83c\udfe5.\n\nIntegrating AI with electronic health records (EHRs) is complex, manual, and time-consuming. Let's try to change that.\n\n\n```bash\npip install healthchain\n```\nFirst time here? Check out our [Docs](https://dotimplement.github.io/HealthChain/) page!\n\n\n## Features\n- [x] \ud83d\udd0c **Gateway**: Connect to multiple EHR systems with [unified API](https://dotimplement.github.io/HealthChain/reference/gateway/gateway/) supporting FHIR, CDS Hooks, and SOAP/CDA protocols (sync / async support)\n- [x] \ud83d\udd25 **Pipelines**: Build FHIR-native ML workflows or use [pre-built ones](https://dotimplement.github.io/HealthChain/reference/pipeline/pipeline/#prebuilt) for your healthcare NLP and AI tasks\n- [x] \ud83d\udd04 **InteropEngine**: Convert between FHIR, CDA, and HL7v2 with a [template-based engine](https://dotimplement.github.io/HealthChain/reference/interop/interop/)\n- [x] \ud83d\udd12 Type-safe healthcare data with full type hints and Pydantic validation for [FHIR resources](https://dotimplement.github.io/HealthChain/reference/utilities/fhir_helpers/)\n- [x] \u26a1 Built-in event-driven logging and operation tracking for [audit trails](https://dotimplement.github.io/HealthChain/reference/gateway/events/)\n- [x] \ud83d\ude80 Deploy production-ready applications with [HealthChainAPI](https://dotimplement.github.io/HealthChain/reference/gateway/api/) and FastAPI integration\n- [x] \ud83e\uddea Generate [synthetic healthcare data](https://dotimplement.github.io/HealthChain/reference/utilities/data_generator/) and [sandbox testing](https://dotimplement.github.io/HealthChain/reference/sandbox/sandbox/) utilities\n- [x] \ud83d\udda5\ufe0f Bootstrap configurations with CLI tools\n\n## Why use HealthChain?\n- **EHR integrations are manual and time-consuming** - **HealthChainAPI** abstracts away complexities so you can focus on AI development, not learning FHIR APIs, CDS Hooks, and authentication schemes.\n- **Healthcare data is fragmented and complex** - **InteropEngine** handles the conversion between FHIR, CDA, and HL7v2 so you don't have to become an expert in healthcare data standards.\n- [**Most healthcare data is unstructured**](https://pmc.ncbi.nlm.nih.gov/articles/PMC10566734/) - HealthChain **Pipelines** are optimized for real-time AI and NLP applications that deal with realistic healthcare data.\n- **Built by health tech developers, for health tech developers** - HealthChain is tech stack agnostic, modular, and easily extensible with built-in compliance and audit features.\n\n## HealthChainAPI\n\nThe HealthChainAPI provides a secure integration layer that coordinates multiple healthcare systems in a single application.\n\n### Multi-Protocol Support\n\nConnect to multiple healthcare data sources and protocols:\n\n```python\nfrom healthchain.gateway import (\n HealthChainAPI, FHIRGateway,\n CDSHooksService, NoteReaderService\n)\n\n# Create your healthcare application\napp = HealthChainAPI(\n title=\"My Healthcare AI App\",\n description=\"AI-powered patient care platform\"\n)\n\n# FHIR for patient data from multiple EHRs\nfhir = FHIRGateway()\nfhir.add_source(\"epic\", \"fhir://fhir.epic.com/r4?client_id=...\")\nfhir.add_source(\"medplum\", \"fhir://api.medplum.com/fhir/R4/?client_id=...\")\n\n# CDS Hooks for real-time clinical decision support\ncds = CDSHooksService()\n\n@cds.hook(\"patient-view\", id=\"allergy-alerts\")\ndef check_allergies(request):\n # Your AI logic here\n return {\"cards\": [...]}\n\n# SOAP for clinical document processing\nnotes = NoteReaderService()\n\n@notes.method(\"ProcessDocument\")\ndef process_note(request):\n # Your NLP pipeline here\n return processed_document\n\n# Register everything\napp.register_gateway(fhir)\napp.register_service(cds)\napp.register_service(notes)\n\n# Your API now handles:\n# /fhir/* - Patient data, observations, etc.\n# /cds/* - Real-time clinical alerts\n# /soap/* - Clinical document processing\n\n# Deploy with uvicorn\nif __name__ == \"__main__\":\n import uvicorn\n uvicorn.run(app, port=8888)\n```\n\n### FHIR Operations with AI Enhancement\n\n```python\nfrom healthchain.gateway import FHIRGateway\nfrom fhir.resources.patient import Patient\n\ngateway = FHIRGateway()\ngateway.add_source(\"epic\", \"fhir://fhir.epic.com/r4?...\")\n\n# Add AI transformations to FHIR data\n@gateway.transform(Patient)\ndef enhance_patient(id: str, source: str = None) -> Patient:\n patient = gateway.read(Patient, id, source)\n\n # Get lab results and process with AI\n lab_results = gateway.search(\n Observation,\n {\"patient\": id, \"category\": \"laboratory\"},\n source\n )\n insights = nlp_pipeline.process(patient, lab_results)\n\n # Add AI summary to patient record\n patient.extension = patient.extension or []\n patient.extension.append({\n \"url\": \"http://healthchain.org/fhir/summary\",\n \"valueString\": insights.summary\n })\n\n # Update the patient record\n gateway.update(patient, source)\n return patient\n\n# Automatically available at: GET /fhir/transform/Patient/123?source=epic\n```\n\n## Pipeline\nPipelines provide a flexible way to build and manage processing pipelines for NLP and ML tasks that can easily integrate with complex healthcare systems.\n\n### Building a pipeline\n\n```python\nfrom healthchain.io.containers import Document\nfrom healthchain.pipeline import Pipeline\nfrom healthchain.pipeline.components import (\n TextPreProcessor,\n SpacyNLP,\n TextPostProcessor,\n)\n\n# Initialize the pipeline\nnlp_pipeline = Pipeline[Document]()\n\n# Add TextPreProcessor component\npreprocessor = TextPreProcessor()\nnlp_pipeline.add_node(preprocessor)\n\n# Add Model component (assuming we have a pre-trained model)\nspacy_nlp = SpacyNLP.from_model_id(\"en_core_sci_sm\")\nnlp_pipeline.add_node(spacy_nlp)\n\n# Add TextPostProcessor component\npostprocessor = TextPostProcessor(\n postcoordination_lookup={\n \"heart attack\": \"myocardial infarction\",\n \"high blood pressure\": \"hypertension\",\n }\n)\nnlp_pipeline.add_node(postprocessor)\n\n# Build the pipeline\nnlp = nlp_pipeline.build()\n\n# Use the pipeline\nresult = nlp(Document(\"Patient has a history of heart attack and high blood pressure.\"))\n\nprint(f\"Entities: {result.nlp.get_entities()}\")\n```\n\n#### Working with healthcare data formats\nAdapters handle conversion between healthcare formats (CDA, FHIR) and internal Document objects for seamless EHR integration.\n\n```python\nfrom healthchain.io import CdaAdapter, Document\nfrom healthchain.models import CdaRequest\n\nadapter = CdaAdapter()\n\n# Parse healthcare data into Document\ncda_request = CdaRequest(document=\"<CDA XML content>\")\ndoc = adapter.parse(cda_request)\n\n# Process with your pipeline\nprocessed_doc = nlp_pipeline(doc)\n\n# Access extracted clinical data\nprint(f\"Problems: {processed_doc.fhir.problem_list}\")\nprint(f\"Medications: {processed_doc.fhir.medication_list}\")\n\n# Convert back to healthcare format\nresponse = adapter.format(processed_doc)\n```\n\n### Using pre-built pipelines\nPre-built pipelines are use case specific end-to-end workflows optimized for common healthcare AI tasks.\n\n```python\nfrom healthchain.pipeline import MedicalCodingPipeline\nfrom healthchain.models import CdaRequest\n\n# Load from model ID\npipeline = MedicalCodingPipeline.from_model_id(\n model=\"blaze999/Medical-NER\", task=\"token-classification\", source=\"huggingface\"\n)\n\n# Simple end-to-end processing\ncda_request = CdaRequest(document=\"<CDA XML content>\")\nresponse = pipeline.process_request(cda_request)\n\n# Or manual control for document access\nfrom healthchain.io import CdaAdapter\nadapter = CdaAdapter()\ndoc = adapter.parse(cda_request)\ndoc = pipeline(doc)\n# Access: doc.fhir.problem_list, doc.fhir.medication_list\nresponse = adapter.format(doc)\n```\n\n## Interoperability\n\nThe InteropEngine is a template-based system that allows you to convert between FHIR, CDA, and HL7v2.\n\n```python\nfrom healthchain.interop import create_interop, FormatType\n\nengine = create_interop()\n\nwith open(\"tests/data/test_cda.xml\", \"r\") as f:\n cda_data = f.read()\n\n# Convert CDA to FHIR\nfhir_resources = engine.to_fhir(cda_data, src_format=FormatType.CDA)\n\n# Convert FHIR to CDA\ncda_data = engine.from_fhir(fhir_resources, dest_format=FormatType.CDA)\n```\n\n## Sandbox\n\nTest your AI applications in realistic healthcare contexts with [CDS Hooks](https://cds-hooks.org/) sandbox environments.\n\n```python\nimport healthchain as hc\nfrom healthchain.sandbox.use_cases import ClinicalDecisionSupport\n\n@hc.sandbox\nclass MyCDS(ClinicalDecisionSupport):\n def __init__(self):\n self.pipeline = SummarizationPipeline.from_model_id(\"facebook/bart-large-cnn\")\n\n @hc.ehr(workflow=\"encounter-discharge\")\n def ehr_database_client(self):\n return self.data_generator.generate_prefetch()\n\ncds = MyCDS()\ncds.start_sandbox()\n\n# Run with: healthchain run mycds.py\n```\n\n## Road Map\n- [ ] \ud83d\udd12 Built-in HIPAA compliance validation and PHI detection\n- [ ] \ud83d\udcca Track configurations, data provenance, and monitor model performance with MLFlow integration\n- [ ] \ud83d\ude80 Compliance monitoring, auditing at deployment as a sidecar service\n- [ ] \ud83d\udd04 HL7v2 parsing and FHIR profile conversion support\n- [ ] \ud83e\udde0 Multi-modal pipelines\n\n\n## Contribute\nWe are always eager to hear feedback and suggestions, especially if you are a developer or researcher working with healthcare systems!\n- \ud83d\udca1 Let's chat! [Discord](https://discord.gg/UQC6uAepUz)\n- \ud83d\udee0\ufe0f [Contribution Guidelines](CONTRIBUTING.md)\n\n\n## Acknowledgements \ud83e\udd17\nThis project builds on [fhir.resources](https://github.com/nazrulworld/fhir.resources) and [CDS Hooks](https://cds-hooks.org/) standards developed by [HL7](https://www.hl7.org/) and [Boston Children's Hospital](https://www.childrenshospital.org/).\n\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Python toolkit that makes it easier to connect your AI/ML pipelines to healthcare systems",
"version": "0.10.0",
"project_urls": {
"Documentation": "https://dotimplement.github.io/HealthChain/"
},
"split_keywords": [
"nlp",
" ai",
" llm",
" healthcare",
" ehr",
" mlops"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "1cf3ad4886045d104b03c77ec8a90adc6c417a905571d410b3ef0357661bfbe4",
"md5": "23c95c95f74874052e29721f7be3cfbe",
"sha256": "bf55151f83908aa0e27e26f862eb071b7e1b7a116f5b54fa66e6fe3e81ae73f6"
},
"downloads": -1,
"filename": "healthchain-0.10.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "23c95c95f74874052e29721f7be3cfbe",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<3.12,>=3.9",
"size": 218882,
"upload_time": "2025-08-13T17:38:35",
"upload_time_iso_8601": "2025-08-13T17:38:35.561818Z",
"url": "https://files.pythonhosted.org/packages/1c/f3/ad4886045d104b03c77ec8a90adc6c417a905571d410b3ef0357661bfbe4/healthchain-0.10.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "6f7d6429cbe147a88bbbc25ec5fe823c5175c5959c1f60682d9a28cf58970648",
"md5": "6c91bb06694e7aef89f162fa3130e359",
"sha256": "dc3330793d05d40c20a14e8f9b94cfce67879a20441f221d01b4b0477ff85ab4"
},
"downloads": -1,
"filename": "healthchain-0.10.0.tar.gz",
"has_sig": false,
"md5_digest": "6c91bb06694e7aef89f162fa3130e359",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<3.12,>=3.9",
"size": 154910,
"upload_time": "2025-08-13T17:38:37",
"upload_time_iso_8601": "2025-08-13T17:38:37.086194Z",
"url": "https://files.pythonhosted.org/packages/6f/7d/6429cbe147a88bbbc25ec5fe823c5175c5959c1f60682d9a28cf58970648/healthchain-0.10.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-13 17:38:37",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "healthchain"
}
JSON