Building a FHIR CapabilityStatement for ETL Systems

In clinical data engineering the FHIR CapabilityStatement is frequently mischaracterized as a static compliance artifact. For an ingestion pipeline bridging legacy HL7 v2 feeds with a modern FHIR R4 endpoint, it is a runtime discovery contract: it tells the worker which resources are writable, which search parameters back conditional loads, which profiles a payload must satisfy, and which terminology operations the server actually exposes. Within the FHIR Terminology Server Integration layer of a clinical pipeline, this contract is what lets a worker decide — before it sends a single write — whether $validate and $expand are even available, or whether terminology resolution must fall back to a static snapshot. Skip it and the pipeline runs blind: silent mapping failures, 400 Bad Request cascades during conditional loads, exhausted connection pools, and the audit-trail fragmentation that triggers compliance review. This page shows how to construct the statement, parse it deterministically, and run a startup preflight that fails fast instead of failing silently.

The failure that motivates all of this is concrete. An Airflow-driven job parses HL7 v2 ADT^A08 (patient update) and ORM^O01 (order) messages, extracts PID/PV1/OBX, and issues conditional PUT operations. The target server returns an OperationOutcome bundle: search parameter 'identifier' not supported, profile … not recognized, terminology validation endpoint unreachable for SNOMED CT expansion. The root cause is never the data — it is that the server published a default statement with rest.mode: "client", no searchParam arrays, no supportedProfile, and no $validate/$expand operations, and the worker trusted hardcoded assumptions instead of the advertised contract.

Quick Reference: Required CapabilityStatement Fields for Deterministic ETL

These are the fields an ingestion worker must read (when consuming a server’s statement) or declare (when publishing its own). Anything ambiguous here breaks deterministic execution.

Implementation Pattern

The artifact you publish is a JSON CapabilityStatement; the logic that consumes it is Python. Both matter, and they must agree.

The published statement

A minimal, production-ready statement declaring Patient and Observation with explicit search parameters, profile bindings, conditional interactions, and the terminology operations the pipeline relies on:

{
  "resourceType": "CapabilityStatement",
  "id": "etl-ingestion-contract",
  "url": "https://api.etl-platform.org/fhir/metadata",
  "version": "1.2.0",
  "status": "active",
  "kind": "instance",
  "fhirVersion": "4.0.1",
  "format": ["json"],
  "rest": [
    {
      "mode": "server",
      "security": {
        "cors": true,
        "service": [
          {"coding": [{"system": "http://terminology.hl7.org/CodeSystem/restful-security-service", "code": "OAuth"}]}
        ],
        "description": "Bearer token required. Audit logging enforced on all write operations."
      },
      "resource": [
        {
          "type": "Patient",
          "profile": "http://hl7.org/fhir/StructureDefinition/Patient",
          "supportedProfile": ["https://etl-platform.org/fhir/StructureDefinition/etl-patient-v2"],
          "interaction": [{"code": "read"}, {"code": "search-type"}, {"code": "update"}],
          "conditionalUpdate": true,
          "conditionalCreate": true,
          "searchParam": [
            {"name": "identifier", "type": "token", "definition": "http://hl7.org/fhir/SearchParameter/Patient-identifier"},
            {"name": "active", "type": "boolean"},
            {"name": "birthdate", "type": "date"}
          ]
        },
        {
          "type": "Observation",
          "profile": "http://hl7.org/fhir/StructureDefinition/Observation",
          "supportedProfile": ["https://etl-platform.org/fhir/StructureDefinition/etl-observation-lab"],
          "interaction": [{"code": "create"}, {"code": "search-type"}],
          "searchParam": [
            {"name": "subject", "type": "reference"},
            {"name": "code", "type": "token"},
            {"name": "category", "type": "token"}
          ]
        }
      ],
      "operation": [
        {"name": "validate", "definition": "http://hl7.org/fhir/OperationDefinition/Resource-validate"},
        {"name": "expand", "definition": "http://hl7.org/fhir/OperationDefinition/ValueSet-expand"}
      ]
    }
  ]
}

conditionalUpdate: true lets the worker send PUT with If-None-Exist and avoid duplicate patients on ADT^A08 reprocessing; supportedProfile rejects malformed payloads before they reach persistence; explicit searchParam arrays stop the server falling back to unindexed scans during reconciliation jobs.

The startup preflight in Python

The worker must read the statement once at startup, assert it satisfies the pipeline’s hard requirements, and build a query plan from what it actually advertises — never from hardcoded endpoints. This is the complete, runnable check:

import requests
from dataclasses import dataclass, field

@dataclass(frozen=True)
class CapabilityRequirement:
    """What the ETL job needs the server to guarantee before it runs."""
    fhir_version: str = "4.0.1"
    resources: dict[str, set[str]] = field(default_factory=dict)   # type -> required search params
    conditional: set[str] = field(default_factory=set)             # resource types needing conditionalUpdate
    operations: set[str] = field(default_factory=set)              # required operation names

def fetch_capability_statement(base_url: str, timeout: float = 5.0) -> dict:
    """Discovery call. Never cache a non-200 or a non-active statement."""
    resp = requests.get(
        f"{base_url}/metadata",
        params={"_format": "json"},
        headers={"Accept": "application/fhir+json"},
        timeout=timeout,
    )
    resp.raise_for_status()
    return resp.json()

def preflight(cs: dict, req: CapabilityRequirement) -> list[str]:
    """Return a list of contract violations; empty list == safe to run."""
    problems: list[str] = []

    if cs.get("status") != "active":
        problems.append(f"status is {cs.get('status')!r}, expected 'active'")
    if cs.get("kind") != "instance":
        problems.append(f"kind is {cs.get('kind')!r}, expected 'instance'")
    if cs.get("fhirVersion") != req.fhir_version:
        problems.append(f"fhirVersion {cs.get('fhirVersion')!r} != {req.fhir_version!r}")

    rest = next((r for r in cs.get("rest", []) if r.get("mode") == "server"), None)
    if rest is None:
        return problems + ["no rest entry with mode 'server'"]

    declared_ops = {o["name"] for o in rest.get("operation", [])}
    for op in req.operations - declared_ops:
        problems.append(f"missing operation ${op}")

    by_type = {r["type"]: r for r in rest.get("resource", [])}
    for rtype, needed_params in req.resources.items():
        res = by_type.get(rtype)
        if res is None:
            problems.append(f"resource {rtype} not supported")
            continue
        declared_params = {p["name"] for p in res.get("searchParam", [])}
        for param in needed_params - declared_params:
            problems.append(f"{rtype}: missing searchParam '{param}'")
        if rtype in req.conditional and not res.get("conditionalUpdate"):
            problems.append(f"{rtype}: conditionalUpdate not enabled")

    return problems

# --- wire it up at pipeline startup ---
REQUIREMENT = CapabilityRequirement(
    resources={"Patient": {"identifier", "active"}, "Observation": {"subject", "code"}},
    conditional={"Patient"},
    operations={"validate", "expand"},
)

if __name__ == "__main__":
    cs = fetch_capability_statement("https://api.etl-platform.org/fhir")
    violations = preflight(cs, REQUIREMENT)
    if violations:
        raise SystemExit("CapabilityStatement preflight failed:\n  - " + "\n  - ".join(violations))

The preflight function turns every line of the quick-reference table into an explicit, fail-fast assertion. A worker that calls it at startup converts the entire class of “silent mapping failure” bugs into a single loud crash with an actionable message, before any PHI is touched.

Validation & Testing

Test the preflight against a deliberately broken statement so you prove the negatives, not just the happy path. This runs offline — no live server required — and belongs in CI:

def _server_rest(resources, operations):
    return {"status": "active", "kind": "instance", "fhirVersion": "4.0.1",
            "rest": [{"mode": "server", "resource": resources,
                      "operation": [{"name": n} for n in operations]}]}

# Golden case: fully compliant contract -> zero violations
good = _server_rest(
    resources=[
        {"type": "Patient", "conditionalUpdate": True,
         "searchParam": [{"name": "identifier"}, {"name": "active"}]},
        {"type": "Observation",
         "searchParam": [{"name": "subject"}, {"name": "code"}]},
    ],
    operations=["validate", "expand"],
)
assert preflight(good, REQUIREMENT) == []

# Regression case: the exact production failure -> three named violations
bad = _server_rest(
    resources=[{"type": "Patient", "conditionalUpdate": False,
                "searchParam": [{"name": "active"}]}],
    operations=[],
)
issues = preflight(bad, REQUIREMENT)
assert "Patient: missing searchParam 'identifier'" in issues
assert "Patient: conditionalUpdate not enabled" in issues
assert "missing operation $expand" in issues
assert "resource Observation not supported" in issues
print("preflight tests passed")

For a live server, validate the published JSON itself before deploying it — run it through the $validate operation or the HAPI validator CLI against the FHIR CapabilityStatement StructureDefinition, and confirm GET [base]/metadata?_format=json returns it unchanged. Pin the statement in a distributed cache (Redis/Memcached) with a strict 15–30 minute TTL, and invalidate on 404/410 so a redeploy never serves a stale contract.

Gotchas & Compliance Constraints

Stale cache outlives a server change. The TTL exists precisely because a server can add or drop a search parameter on redeploy. If a worker caches the statement for hours, a removed searchParam produces 400 unknown search parameter long after the contract changed. Keep the TTL tight (15–30 minutes) and force re-discovery on any 404/410 from a previously valid endpoint — never extend the TTL “for performance.”

Terminology operations are part of this contract, not a separate concern. If the statement omits $expand or $validate, the worker must not speculatively call them — that is the fastest route to connection-pool exhaustion. Read the declared operations here and switch the pipeline to a static value-set snapshot fallback, exactly as the FHIR Terminology Server Integration workflow specifies. The same discipline applies to the search parameters your conditional loads depend on; see configuring FHIR search parameters for ETL for how to derive query plans from the declared searchParam arrays.

Data minimization and audit logging are enforced through this document. Declare only the resources and search parameters your ingestion scope actually needs — omitting Observation or Condition when they are out of scope prevents accidental PHI leakage through exploratory queries (HIPAA minimum-necessary, §164.502(b)). Record security.description noting mandatory audit logging, and ensure every create/update/delete interaction is captured in an immutable log per HIPAA §164.312(b). Reject any statement served over plaintext HTTP, and map security.service to least-privilege OAuth2/SMART-on-FHIR scopes so the ETL service account can only touch the resources it declares.

A condensed view of the symptom-to-fix mapping these constraints produce:

Related

• FHIR Terminology Server Integration — the parent topic this discovery contract supports
• Configuring FHIR search parameters for ETL — turning declared searchParam arrays into deterministic query plans
• Validating FHIR resources against US Core profiles — enforcing the supportedProfile declarations at write time
• FHIR & HL7 v2 Standards Architecture for Clinical ETL — the full pipeline architecture