DEV Community

Jack Jiang
Jack Jiang

Posted on

Wiring Up SMART on FHIR OAuth2 Between OpenEMR and a Python Microservice Locally

#python #openemr #oauth2 #fastapi

I was building an AI agent on top of OpenEMR and hit a wall I didn't expect.

Not a hard engineering problem — just documentation. Or the lack of it. OpenEMR ships with a full SMART on FHIR OAuth2 server, and the official authentication docs are a decent starting point. But connecting a custom Python microservice to it? You're mostly on your own.

I spent way more time than I'd like to admit staring at active: false responses, getting HTML back from endpoints that should return JSON, and debugging DNS failures that turned out to be a two-word fix. So I'm writing down everything I wish I'd known — the steps and the gotchas.

The Setup

Here's what we're working with:

Service Tech Port
openemr OpenEMR 7.0.4 (PHP + Docker) 8300 (HTTP), 9300 (HTTPS)
ai-agent Python FastAPI + LangGraph 8000

Both run in Docker Compose. That detail matters more than it sounds — the OpenEMR container is reachable internally at http://openemr, but the browser needs http://localhost:8300. Mixing these up is the single most common source of errors in this whole setup. Keep that distinction in your head throughout.

How OpenEMR's OAuth2 Actually Works

OpenEMR implements SMART on FHIR on top of a standard OAuth2 Authorization Code flow, using the league/oauth2-server library under the hood.

The endpoints you'll actually use (all under /oauth2/default/):

Endpoint Purpose
GET /authorize Redirects the user to OpenEMR login
POST /token Exchanges an authorization code for tokens
POST /introspect Validates a token and returns its claims (RFC 7662)
POST /registration Registers an OAuth client dynamically

There are also two distinct scope families, and this distinction really matters:

  • Patient scopes — patient/Patient.read, patient/Appointment.read, etc. These are tied to a specific authenticated patient.
  • Staff scopes — system/Patient.read, system/Appointment.read, etc. For provider and admin access.

Your microservice sits at the end of this flow. It receives a Bearer token from whatever client is calling it, and it needs to validate that token against OpenEMR before trusting anything about the request.

Step 1: Register an OAuth Client

Before your service can introspect tokens, it has to be a registered client in OpenEMR itself. You have two options.

Option A: The Admin UI

Easiest for a one-time local setup.

  1. Log into OpenEMR at http://localhost:8300 — use HTTP here, not HTTPS on 9300, to avoid a CORS headache
  2. Go to Admin → System → OAuth2 Client Registration
  3. Create a confidential client:
    • Application Type: confidential
    • Redirect URI: your service's callback (e.g. http://localhost:8000/auth/callback)
    • Scopes: include all the patient and system scopes your service needs
    • Grant Types: authorization_code, refresh_token

Save the client_id and client_secret. You'll need both for introspection.

Option B: Dynamic Client Registration

OpenEMR supports RFC 7591, which means you can register clients programmatically. Here's a script that does exactly that and prints the credentials you need:

# scripts/register_oauth_client.py
import sys
import time
import httpx

OPENEMR_BASE = "http://localhost:8300"
REGISTRATION_URL = f"{OPENEMR_BASE}/oauth2/default/registration"

CLIENT_PAYLOAD = {
    "application_type": "confidential",
    "client_name": "My AI Agent",
    "redirect_uris": ["http://localhost:8000/auth/callback"],
    "scope": (
        "openid fhirUser "
        "patient/Patient.read patient/Appointment.read patient/Appointment.write "
        "system/Patient.read system/Appointment.read system/Appointment.write "
        "system/Practitioner.read system/Coverage.read"
    ),
    "grant_types": ["authorization_code", "refresh_token"],
    "response_types": ["code"],
}


def register(payload: dict) -> dict:
    for attempt in range(5):
        try:
            resp = httpx.post(REGISTRATION_URL, json=payload, timeout=15)
            if resp.status_code in (200, 201):
                return resp.json()
            print(f"HTTP {resp.status_code}: {resp.text[:200]}")
        except httpx.ConnectError:
            print(f"OpenEMR not ready, retrying in 5s... (attempt {attempt + 1})")
            time.sleep(5)
    sys.exit(1)


if __name__ == "__main__":
    result = register(CLIENT_PAYLOAD)
    print(f"OPENEMR_CLIENT_ID={result['client_id']}")
    print(f"OPENEMR_CLIENT_SECRET={result['client_secret']}")
Enter fullscreen mode Exit fullscreen mode

Run this once after OpenEMR starts. Copy the output into your .env. Done.

Gotcha: If the dynamic registration endpoint rejects your request, check Admin → Globals → Site → Site Address. The site_addr_oauth value must match the URL you're calling from. When there's a mismatch, OpenEMR quietly rejects registration requests — no helpful error message, just a failure.

Step 2: Environment Variables 

# .env

# Set after running scripts/register_oauth_client.py
OPENEMR_CLIENT_ID=
OPENEMR_CLIENT_SECRET=

# Internal Docker URL for server-to-server calls
OPENEMR_TOKEN_URL=http://openemr/oauth2/default/token
Enter fullscreen mode Exit fullscreen mode

That OPENEMR_TOKEN_URL value — http://openemr — is the internal Docker hostname. It only resolves from inside the Docker network. If you're running your agent locally with uvicorn outside of Docker, change it to http://localhost:8300. I cannot stress enough how many confusing errors trace back to this one hostname distinction.

Step 3: The Token Introspection Service

Here's the core of the whole thing. On every incoming request, your service needs to call OpenEMR's /introspect endpoint with the Bearer token, and OpenEMR will tell you whether it's valid and what it's allowed to do.

The response looks like this:

{ "active": true, "sub": "some-user-uuid", "scope": "patient/Patient.read patient/Appointment.read" }
Enter fullscreen mode Exit fullscreen mode

You take that, check the scopes, and either proceed or reject the request. Simple in theory — but there are details worth getting right:

# app/services/auth_service.py
"""
Token validation via OpenEMR OAuth2 introspection (RFC 7662).
Caches results briefly to avoid hammering OpenEMR on every request.
"""

import hashlib
import logging
import os
import time

import httpx

logger = logging.getLogger(__name__)

_INTROSPECT_CACHE_TTL = 45  # seconds — shorter than token lifetime
_cache: dict[str, tuple[dict, float]] = {}


def _introspect_url() -> str:
    token_url = os.getenv("OPENEMR_TOKEN_URL", "http://openemr/oauth2/default/token")
    return token_url.rstrip("/").replace("/token", "") + "/introspect"


def _cache_key(token: str) -> str:
    # Hash the token so raw credentials never appear in memory keys
    return hashlib.sha256(token.encode()).hexdigest()[:32]


def _get_cached(token: str) -> dict | None:
    key = _cache_key(token)
    entry = _cache.get(key)
    if not entry:
        return None
    claims, expires_at = entry
    if time.time() >= expires_at:
        del _cache[key]
        return None
    return claims


def _set_cached(token: str, claims: dict) -> None:
    key = _cache_key(token)
    _cache[key] = (claims, time.time() + _INTROSPECT_CACHE_TTL)


async def _introspect(token: str) -> dict:
    """Call OpenEMR introspection endpoint."""
    url = _introspect_url()
    client_id = os.getenv("OPENEMR_CLIENT_ID", "")
    client_secret = os.getenv("OPENEMR_CLIENT_SECRET", "")

    if not client_id or not client_secret:
        logger.warning("Introspection skipped: OPENEMR_CLIENT_ID/SECRET not configured")
        return {"active": False}

    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.post(
            url,
            data={
                "token": token,
                "token_type_hint": "access_token",
                "client_id": client_id,
                "client_secret": client_secret,
            },
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )

    if resp.status_code != 200:
        logger.warning("Introspection HTTP %s", resp.status_code)
        return {"active": False}

    try:
        return resp.json()
    except Exception as exc:
        logger.warning("Introspection parse error: %s", exc)
        return {"active": False}


async def validate_patient_token(token: str) -> dict:
    """
    Validate a patient token. Raises HTTPException if invalid.
    Returns introspection claims on success (includes 'sub' = patient FHIR ID).
    """
    from fastapi import HTTPException

    cached = _get_cached(token)
    claims = cached if cached is not None else await _introspect(token)

    if cached is None:
        _set_cached(token, claims)

    if not claims.get("active"):
        logger.warning("Patient token invalid: prefix=%s", token[:8])
        raise HTTPException(status_code=401, detail="Token is invalid or expired")

    scopes = (claims.get("scope") or "").split()
    if not any(s.startswith("patient/") for s in scopes):
        logger.warning("Patient token lacks patient scope: prefix=%s", token[:8])
        raise HTTPException(status_code=403, detail="Token lacks patient scope")

    if not claims.get("sub"):
        logger.warning("Patient token missing sub claim: prefix=%s", token[:8])
        raise HTTPException(status_code=401, detail="Token missing subject claim")

    return claims


async def validate_staff_token(token: str) -> dict:
    """
    Validate a staff token. Raises HTTPException if invalid.
    Returns introspection claims on success.
    """
    from fastapi import HTTPException

    cached = _get_cached(token)
    claims = cached if cached is not None else await _introspect(token)

    if cached is None:
        _set_cached(token, claims)

    if not claims.get("active"):
        logger.warning("Staff token invalid: prefix=%s", token[:8])
        raise HTTPException(status_code=401, detail="Token is invalid or expired")

    scopes = (claims.get("scope") or "").split()
    if not any(s.startswith("system/") for s in scopes):
        logger.warning("Staff token lacks system scope: prefix=%s", token[:8])
        raise HTTPException(status_code=403, detail="Token lacks staff/system scope")

    return claims
Enter fullscreen mode Exit fullscreen mode

A few decisions in here that are worth calling out:

  • The TTL cache. Calling introspect on every single request would hammer OpenEMR. A 45-second in-memory cache keeps things fast without risking stale results — access tokens typically live for an hour. Tokens are hashed before use as cache keys so the raw strings never end up sitting in memory.
  • 403 vs 401. This distinction matters. A missing or expired token is a 401 — unauthenticated. A valid staff token presented to a patient endpoint is a 403 — authenticated but wrong scope. Conflating these makes debugging a nightmare.
  • Log the prefix, not the token. First 8 characters is enough to correlate a failure to a specific token in your logs without actually leaking credentials.

Step 4: Wiring It Into FastAPI

With the validation service built, the endpoint itself stays clean. FastAPI's HTTPBearer dependency pulls the token from the Authorization header, and validate_patient_token handles everything else:

# app/api/patient.py
from fastapi import APIRouter, Depends
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from app.services.auth_service import validate_patient_token

router = APIRouter(prefix="/api/chat", tags=["patient"])
security = HTTPBearer()


def _get_patient_token(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> str:
    return credentials.credentials


@router.post("/patient", response_model=ChatResponse)
async def patient_chat(
    body: ChatRequest,
    token: str = Depends(_get_patient_token),
) -> ChatResponse:
    token_claims = await validate_patient_token(token)  # raises 401/403 if bad

    # Use the verified patient ID from the token for FHIR data scoping
    patient_id = str(token_claims["sub"])

    # Pass the validated token downstream so FHIR calls are made on behalf of this patient
    message, tool_calls, usage, metadata = await asyncio.to_thread(
        invoke_patient_agent,
        user_input,
        patient_id=patient_id,
        fhir_token=token,
    )
    ...
Enter fullscreen mode Exit fullscreen mode

token_claims["sub"] is the patient's identity. Passing fhir_token to the agent means every downstream FHIR call is made on behalf of that specific patient — which is the whole point of SMART on FHIR's data isolation model.

The Pitfalls I Hit (So You Don't Have To)

Introspection returns active: false on a perfectly fresh token.
Nine times out of ten this means your OPENEMR_CLIENT_ID and OPENEMR_CLIENT_SECRET don't match a registered confidential client. The introspecting client must itself be registered in OpenEMR — you can't just use arbitrary credentials.

Getting HTML back instead of JSON from the introspect endpoint.
This one is deeply confusing the first time. It means the site address (site_addr_oauth) in OpenEMR's globals doesn't match the issuer URL. OpenEMR redirects to an error page instead of returning JSON, and if you're not printing the raw response, all you see is a parse error. Fix it at Admin → Config → Connectors → Site Address Override.

DNS failure when calling http://openemr/....
You're running the agent outside Docker. The openemr hostname only resolves inside the Docker network. Switch to http://localhost:8300 for local development.

Token validates but the patient ID looks wrong.
The sub claim from introspection is the internal OpenEMR user UUID — not the FHIR patient resource ID. If you need the FHIR patient ID directly, cross-reference it via the FHIR API, or grab the patient claim from the original token response (only present for patient-context EHR launches).

Introspection passes but the FHIR API still returns 401.
The FHIR API and the introspection endpoint are independently authorized. A token can be active: true but still get rejected by a specific FHIR endpoint if the token's scopes don't cover that resource type. Make sure the scopes registered on your client actually match what your queries need.

Takeaways

  • OpenEMR's OAuth2 is real OAuth2 — once you know where the gotchas are, it behaves like a standard SMART on FHIR server. The pain is mostly in the setup, not the protocol.
  • The internal vs. external hostname split will bite you. Keep http://openemr for server-to-server calls inside Docker, http://localhost:8300 for everything else. Tattoo it on your brain.
  • Introspection is the right tool. Your service shouldn't be verifying JWT signatures against OpenEMR's keys — use /introspect and let OpenEMR do that work. Cache the results and you won't pay a meaningful performance cost.
  • Scope-based separation makes roles explicit. Patient endpoints enforce patient/* scopes, staff endpoints enforce system/* scopes. A 403 tells you who the person is but why they can't access this — that's a much more useful error than a generic 401.

If you're building anything on OpenEMR and run into a different gotcha, drop it in the comments. This stuff is genuinely undertested territory and the more people document it, the better.

Top comments (0)