Gateway

The HealthChain Gateway module provides a secure, asynchronous integration layer for connecting your NLP/ML pipelines with multiple healthcare systems. It provides a unified interface for connecting to FHIR servers, CDS Hooks, and SOAP/CDA services and is designed to be used in conjunction with the HealthChainAPI to create a complete healthcare integration platform.

Features 🚀

The Gateway handles the complex parts of healthcare integration:

• Multiple Protocols: Works with FHIR RESTful APIs, CDS Hooks, and Epic NoteReader CDI (SOAP/CDA service) out of the box
• Multi-Source: Context managers to work with data from multiple EHR systems and FHIR servers safely
• Smart Connections: Handles OAuth2.0 authentication, connection pooling, and automatic token refresh
• Event-Driven: Native asyncio support for real-time events, audit trails, and workflow automation
• Transform & Aggregate: FastAPI-style declarative patterns to create endpoints for enhancing and combining data
• Developer-Friendly: Modern Python typing and validation support via fhir.resources (powered by Pydantic), protocol-based interfaces, and informative error messages

Key Components

Component	Description	Use Case
HealthChainAPI	FastAPI app with gateway and service registration	Main app that coordinates everything
FHIRGateway	FHIR client with connection pooling and authentication	Reading/writing patient data from EHRs (Epic, Cerner, etc.) or application FHIR servers (Medplum, Hapi etc.)
CDSHooksService	Clinical Decision Support hooks service	Real-time alerts and recommendations
NoteReaderService	SOAP/CDA document processing service	Processing clinical documents and notes
Event System	Event-driven integration	Audit trails, workflow automation

Basic Usage

from healthchain.gateway import HealthChainAPI, FHIRGateway
from fhir.resources.patient import Patient

# Create the application
app = HealthChainAPI()

# Create and configure a FHIR gateway
fhir = FHIRGateway()

# Connect to your FHIR APIs
fhir.add_source("epic", "fhir://epic.org/api/FHIR/R4?client_id=...")
fhir.add_source("medplum", "fhir://api.medplum.com/fhir/R4/?client_id=...")

# Add AI enhancements to patient data
@fhir.transform(Patient)
async def enhance_patient(id: str, source: str = None) -> Patient:
    async with fhir.modify(Patient, id, source) as patient:
        patient.active = True  # Your custom logic here
        return patient

# Register and run
app.register_gateway(fhir)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app)

# Default: http://127.0.0.1:8000/

You can also register multiple services of different protocols!

from healthchain.gateway import (
    HealthChainAPI, FHIRGateway,
    CDSHooksService, NoteReaderService
)

app = HealthChainAPI()

# FHIR for patient data
fhir = FHIRGateway()
fhir.add_source("epic", "fhir://fhir.epic.com/r4?...")

# CDS Hooks for real-time alerts
cds = CDSHooksService()

@cds.hook("patient-view", id="allergy-alerts")
def check_allergies(request):
    # Your logic here
    return {"cards": [...]}

# SOAP for clinical documents
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)

Protocol Support

Protocol	Implementation	Features
FHIR API	FHIRGateway	FHIR-instance level CRUD operations - read, create, update, delete, search, register transform and aggregate handlers, connection pooling and authentication management
CDS Hooks	CDSHooksService	Hook Registration, Service Discovery
SOAP/CDA	NoteReaderService	Method Registration (ProcessDocument), SOAP Service Discovery (WSDL)