IntaOps Connector — Python

Drop-in Python SDK for the IntaOps SMART-on-FHIR R4 network.

pip install intaops

Prerequisites

Before the SDK can call IntaOps, you need OAuth credentials — client_id + client_secret. Get them from the IntaOps Dashboard:

1. Sign in → Developer Portal → Applications (https://intaops.io/applications)
2. Create an Application — pick client_credentials for backend services, or authorization_code for provider / patient-facing apps
3. Copy the client_secret when shown — it's revealed once. Keep it server-side, never in browser code.

The SDK handles the OAuth token exchange (POST /oauth/token) for you — IntaOpsClient(client_id=…, client_secret=…) is enough. Access tokens are short-lived (~5 min) and refreshed automatically.

Backend (system-to-system)

from intaops import IntaOpsClient

client = IntaOpsClient(
    client_id="app_live_…",
    client_secret="secret_live_…",
    environment="live",                  # or "sandbox"
    scopes=["system/Patient.read", "system/Observation.read"],
)
patient = client.fhir.get("Patient", "u123")
for obs in client.fhir.search_iter("Observation", patient="u123"):
    print(obs["valueQuantity"]["value"])

Provider / patient app (SMART user grant)

from intaops import IntaOpsClient

# Step 1 — generate PKCE + redirect the user
pkce = IntaOpsClient.generate_pkce()
client = IntaOpsClient(
    client_id="app_live_…", environment="live",
    scopes=["patient/Observation.read", "patient/MedicationRequest.read"],
)
url = client.authorize_url(
    redirect_uri="https://yourapp.com/cb",
    code_challenge=pkce.challenge,
)
# … redirect the user to `url`, capture `code` on the callback …

# Step 2 — exchange the code for tokens
client = IntaOpsClient.from_authorization_code(
    code=code_from_callback,
    redirect_uri="https://yourapp.com/cb",
    code_verifier=pkce.verifier,
    client_id="app_live_…",
    environment="live",
    scopes=["patient/Observation.read", "patient/MedicationRequest.read"],
)
obs = client.fhir.get("Observation", "obs-1")

Errors

from intaops import (
    IntaOpsError, OAuthError, OperationOutcomeError,
    ConsentRequiredError, RateLimitedError,
)

try:
    client.fhir.get("Observation", "obs-1")
except ConsentRequiredError as e:
    # First call to any patient/* scope ALWAYS returns this — consent is OFF
    # by default. The IntaOps backend auto-pushes a prompt to the individual's
    # IntaOps app; surface e.consent_url to them (push, email, in-app deep-link)
    # and retry once they approve.
    notify_user(e.user_id, e.consent_url)
except RateLimitedError as e:
    time.sleep(e.retry_after or 30)
except OperationOutcomeError as e:
    log(e.status_code, e.issue_code, e.location, e.outcome)
except OAuthError as e:
    refresh_credentials(e.error)

Scope ↔ industry gating — the scopes your app can request are bounded by your entity's industry. Healthcare entities can request patient/* (consent-gated) and system/* (broad read) FHIR scopes; other industries are scoped to their own data surfaces (financial → financial data, retail → shopping, etc.). Requesting a scope outside your industry raises OAuthError("invalid_scope") at client instantiation.

API

• IntaOpsClient(client_id, client_secret=, environment=, scopes=)
• IntaOpsClient.from_authorization_code(code, redirect_uri, code_verifier, …)
• IntaOpsClient.generate_pkce() → PKCEPair
• IntaOpsClient.authorize_url(redirect_uri, code_challenge, state=) → str
• client.fhir.get(resource_type, resource_id) → dict
• client.fhir.search(resource_type, **params) → Bundle dict
• client.fhir.search_iter(resource_type, **params) → iterator of resources
• client.fhir.create(resource_type, resource) → dict
• client.fhir.update(resource_type, resource_id, resource) → dict
• client.fhir.capability_statement() → dict

Marketplace & interoperability — client.marketplace

Also importable standalone: from intaops.marketplace import MarketplaceClient.

• client.marketplace.products.create(**data) · .create_with_transform(**data) · .update(product_id, **data) · .get(product_id) · .search(**params)
• client.marketplace.products.bulk_import(products) · .csv_upload(file, filename="products.csv")
• client.marketplace.payments.initiate(**data) · .process(transaction_id) · .history(**params) · .revenue()
• client.marketplace.channels.create(**data) · .send(channel_id, **data)
• client.marketplace.webhooks.subscribe(**data) · .list() · .update(subscription_id, **data) · .delete(subscription_id) · .deliveries(subscription_id) · .test(subscription_id)
• client.marketplace.transform.create_schema_mapping(**data)
• MarketplaceClient.verify_signature(payload, signature, secret) → bool — verify the Intaops-Signature header on inbound webhooks

Migration (EHR / EMR import) — client.migration

Also importable standalone: from intaops.migration import MigrationClient.

• client.migration.initiate(...) → job dict
• client.migration.status(job_id) · .jobs(limit=50)
• client.migration.push_records(job_id, records)
• client.migration.fhir_import(**kwargs) · .fhir_import_status(job_id)

License

Apache-2.0.