Building Your Own Epic FHIR Integration: A Complete Guide

import jwt

import time

import uuid

from datetime import datetime, timedelta

def create_jwt(client_id, private_key_path, token_url):

    """

    Create a signed JWT for Epic authentication.

    Args:

        client_id: Your Epic client ID

        private_key_path: Path to your RSA private key

        token_url: Epic's token endpoint

    Returns:

        Signed JWT string

    """

    # Read private key

    with open(private_key_path, 'r') as key_file:

        private_key = key_file.read()  

    # JWT claims

    now = int(time.time())

    claims = {

        'iss': client_id,  # Issuer: your client ID

        'sub': client_id,  # Subject: also your client ID

        'aud': token_url,  # Audience: Epic's token endpoint

        'jti': str(uuid.uuid4()),  # Unique identifier for this JWT

        'exp': now + 300,  # Expiration: 5 minutes from now

        'iat': now,  # Issued at: current time

    }   

    # Sign with RS384 algorithm (required by Epic)

    signed_jwt = jwt.encode(claims, private_key, algorithm='RS384')

    return signed_jwt

import aiohttp

async def get_access_token(client_id, private_key_path, token_url):

    """

    Exchange JWT for an access token.   

    Args:

        client_id: Your Epic client ID

        private_key_path: Path to your RSA private key

        token_url: Epic's token endpoint    

    Returns:

        Access token string

    """

    # Create JWT

    jwt_token = create_jwt(client_id, private_key_path, token_url)   

    # Prepare request

    data = {

        'grant_type': 'client_credentials',

        'client_assertion_type': 'urn:ietf:params:oauth:client-assertion-type:jwt-bearer',

        'client_assertion': jwt_token,

    }    

    # Exchange JWT for access token

    async with aiohttp.ClientSession() as session:

        async with session.post(token_url, data=data) as response:

            if response.status != 200:

                error_text = await response.text()

                raise Exception(f"Authentication failed: {error_text}")          

            token_data = await response.json()

            return token_data['access_token']

from datetime import datetime, timedelta

class TokenCache:

    """Simple token cache with expiration."""    

    def __init__(self):

        self.token = None

        self.expires_at = None    

    def get_token(self):

        """Get cached token if still valid."""

        if self.token and self.expires_at > datetime.now():

            return self.token

        return None    

    def set_token(self, token, expires_in_seconds=3300):

        """Cache token with expiration (default 55 minutes)."""

        self.token = token

        self.expires_at = datetime.now() + timedelta(seconds=expires_in_seconds)    

    def invalidate(self):

        """Invalidate cached token."""

        self.token = None

        self.expires_at = Non

async def get_patient(patient_id, access_token, base_url):

    """

    Retrieve a patient by ID.    

    Args:

        patient_id: Epic patient ID

        access_token: Valid access token

        base_url: Epic FHIR base URL    

    Returns:

        Patient resource as dictionary

    """

    url = f"{base_url}/Patient/{patient_id}"    

    headers = {

        'Authorization': f'Bearer {access_token}',

        'Accept': 'application/fhir+json',

    }    

    async with aiohttp.ClientSession() as session:

        async with session.get(url, headers=headers) as response:

            if response.status == 200:

                return await response.json()

            else:

                error_text = await response.text()

                raise Exception(f"Request failed: HTTP {response.status} - {error_text}")

async def search_observations(patient_id, category, access_token, base_url):

    """

    Search for observations by patient and category.

    Args:

        patient_id: Epic patient ID

        category: Observation category (e.g., 'vital-signs', 'laboratory')

        access_token: Valid access token

        base_url: Epic FHIR base URL

    Returns:

        Bundle of observations

    """

    url = f"{base_url}/Observation"

    params = {

        'patient': patient_id,

        'category': category,

        '_count': 20,  # Results per page

        '_sort': '-date',  # Sort by date descending

    }

    headers = {

        'Authorization': f'Bearer {access_token}',

        'Accept': 'application/fhir+json',

    }

    async with aiohttp.ClientSession() as session:

        async with session.get(url, headers=headers, params=params) as response:

            if response.status == 200:

                return await response.json()

            else:

                error_text = await response.text()

                raise Exception(f"Search failed: {error_text}")