Canvas FHIR

The Canvas SDK FHIR client provides a simple interface for interacting with the Canvas FHIR API, supporting CRUD operations on FHIR resources such as Coverages, DocumentReferences, AllergyIntolerances, and more. It handles OAuth client credentials authentication and token caching automatically.

Requirements #

  • Canvas FHIR Client ID: An OAuth client ID for your Canvas environment
  • Canvas FHIR Client Secret: The corresponding OAuth client secret

These credentials should be stored as plugin secrets and grant access to the Canvas FHIR API for your environment.

Imports #

The Canvas FHIR client is included in the Canvas SDK. Import the client:

from canvas_sdk.clients.canvas_fhir import CanvasFhir

Initialize the Client #

# Declare these secrets in the CANVAS_MANIFEST.json and set the values on the
# plugin configuration page.
client_id = self.secrets["CANVAS_FHIR_CLIENT_ID"]
client_secret = self.secrets["CANVAS_FHIR_CLIENT_SECRET"]

client = CanvasFhir(client_id, client_secret)

On initialization, the client will:

  1. Authenticate using the OAuth client credentials flow against your Canvas environment’s token endpoint.
  2. Cache the access token using the plugin cache system, keyed by client_id, with automatic expiration.
  3. Determine the FHIR API base URL from the environment’s CUSTOMER_IDENTIFIER setting (e.g., https://fumage-{CUSTOMER_IDENTIFIER}.canvasmedical.com).

CanvasFhir #

The main class for interacting with the Canvas FHIR API.

Constructor #

CanvasFhir(client_id: str, client_secret: str)
ParameterTypeDescription
client_idstrOAuth client ID for the Canvas API
client_secretstrOAuth client secret

Methods #

search(resource_type: str, parameters: dict) -> dict #

Search for FHIR resources matching the given parameters.

# Search for a patient's allergy intolerances
results = client.search("AllergyIntolerance", {"patient": "Patient/abc123"})

for entry in results.get("entry", []):
    resource = entry["resource"]
    print(f"Allergy: {resource['code']['coding'][0]['display']}")
ParameterTypeDescription
resource_typestrFHIR resource type (e.g., Patient, Coverage)
parametersdictSearch parameters as key-value pairs

Returns: FHIR Bundle dict containing matching resources.

Raises: requests.HTTPError if the API returns an error status code.

read(resource_type: str, resource_id: str) -> dict #

Read a single FHIR resource by its ID.

# Read a specific resource by ID
allergy = client.read("AllergyIntolerance", "allergy-id-123")
print(f"Status: {allergy['clinicalStatus']['coding'][0]['code']}")
ParameterTypeDescription
resource_typestrFHIR resource type
resource_idstrID of the resource to read

Returns: FHIR resource dict.

Raises: requests.HTTPError if the API returns an error status code.

create(resource_type: str, data: dict) -> dict #

Create a new FHIR resource.

# Create a new Coverage resource
coverage = client.create("Coverage", {
    "resourceType": "Coverage",
    "status": "active",
    "beneficiary": {"reference": "Patient/abc123"},
    "payor": [{"reference": "Organization/org-456"}],
})
print(f"Created Coverage: {coverage['id']}")
ParameterTypeDescription
resource_typestrFHIR resource type
datadictFHIR resource data to create

Returns: Created FHIR resource dict (including server-assigned id).

Raises: requests.HTTPError if the API returns an error status code.

update(resource_type: str, resource_id: str, data: dict) -> dict #

Update an existing FHIR resource.

# Update an existing resource
updated = client.update("Coverage", "coverage-id-789", {
    "resourceType": "Coverage",
    "id": "coverage-id-789",
    "status": "cancelled",
    "beneficiary": {"reference": "Patient/abc123"},
    "payor": [{"reference": "Organization/org-456"}],
})
print(f"Updated Coverage status: {updated['status']}")
ParameterTypeDescription
resource_typestrFHIR resource type
resource_idstrID of the resource to update
datadictComplete FHIR resource data

Returns: Updated FHIR resource dict.

Raises: requests.HTTPError if the API returns an error status code.

Authentication #

The client uses the OAuth 2.0 client credentials flow to authenticate with the Canvas API. Token management is handled automatically:

  • On first use, the client exchanges the client_id and client_secret for an access token via the Canvas token endpoint.
  • The token is cached using the plugin cache system with the key canvas_fhir_credentials_{client_id}.
  • The cached token expires 60 seconds before the actual token expiration to avoid using stale credentials.
  • Subsequent requests reuse the cached token until it expires.

Error Handling #

The Canvas FHIR client uses raise_for_status() on all HTTP responses, which raises requests.HTTPError for non-successful status codes.

from requests import HTTPError

try:
    result = client.read("Patient", "nonexistent-id")
except HTTPError as e:
    print(f"HTTP {e.response.status_code}: {e.response.text}")

Complete Plugin Example #

Here’s a complete example of using the Canvas FHIR client in an ActionButton handler:

from canvas_sdk.clients.canvas_fhir import CanvasFhir
from canvas_sdk.effects import Effect
from canvas_sdk.handlers.action_button import ActionButton
from logger import log


class FhirRequestHandler(ActionButton):
    """Handler that queries the FHIR API when a button is clicked."""

    BUTTON_TITLE = "Trigger FHIR Request"
    BUTTON_KEY = "TRIGGER_FHIR_REQUEST"
    BUTTON_LOCATION = ActionButton.ButtonLocation.CHART_SUMMARY_ALLERGIES_SECTION

    def handle(self) -> list[Effect]:
        """Handle the button click."""
        client_id = self.secrets["CANVAS_FHIR_CLIENT_ID"]
        client_secret = self.secrets["CANVAS_FHIR_CLIENT_SECRET"]
        patient_id = self.event.target.id

        client = CanvasFhir(client_id, client_secret)

        # Search for the patient's allergy intolerances
        search_response = client.search(
            "AllergyIntolerance",
            {"patient": f"Patient/{patient_id}"},
        )
        log.info(f"Search: {search_response}")

        # Read the first result
        first_entry = search_response["entry"][0]["resource"]
        read_response = client.read("AllergyIntolerance", first_entry["id"])
        log.info(f"Read: {read_response}")

        return []

Additional Resources #




Questions? Create an issue.
On this page