Is there a way to get to "Data Integration" files from SDK?

[nara-falconeer]

To process incoming files by fax, should I just go to Documo and pull up the files or does the SDK have an interface for this? I checked the IntegrationTask (https://docs.canvasmedical.com/sdk/data-integration-task/) but there is no DocumentReference or such in it.

[rmagier1]

Hi @nara-falconeer — good question. The IntegrationTask data class itself doesn't carry the file, but you don't need to go to Documo for this. Two paths depending on when you want to act:

For automating inbound-fax intake (most common case) — hook the DOCUMENT_RECEIVED event. It fires the moment a new IntegrationTask is created, regardless of channel (fax, document upload, lab, etc.), and the event context includes the file URL directly. Here's a minimal handler that just logs the payload so you can see exactly what shows up — drop it into a plugin, install it, and trigger an inbound document:

CANVAS_MANIFEST.json:

{
  "sdk_version": "0.1.4",
  "plugin_version": "0.0.1",
  "name": "document_received_logger",
  "description": "Logs DOCUMENT_RECEIVED event context for inspection.",
  "components": {
    "protocols": [
      {
        "class": "document_received_logger.handlers.document_received:DocumentReceivedLogger",
        "description": "Logs the DOCUMENT_RECEIVED event context."
      }
    ],
    "commands": [], "content": [], "effects": [],
    "views": [], "questionnaires": []
  },
  "secrets": [], "tags": {}, "references": [], "license": "",
  "diagram": false, "readme": "./README.md"
}

document_received_logger/handlers/document_received.py:

import json

from canvas_sdk.effects import Effect
from canvas_sdk.events import EventType
from canvas_sdk.handlers import BaseHandler
from logger import log


class DocumentReceivedLogger(BaseHandler):
    """Log every DOCUMENT_RECEIVED event for inspection."""

    RESPONDS_TO = EventType.Name(EventType.DOCUMENT_RECEIVED)

    def compute(self) -> list[Effect]:
        ctx = self.context or {}
        document = ctx.get("document") or {}
        patient = ctx.get("patient") or None
        available_types = ctx.get("available_document_types") or []

        # Use the IntegrationTask UUID as the correlation id so you can grep
        # across DOCUMENT_RECEIVED, DOCUMENT_LINKED_TO_PATIENT, DOCUMENT_REVIEWED,
        # etc. for the same task with a single search.
        cid = getattr(self.event, "target", None) or document.get("id")

        log.info(
            f"[doc_received] short cid={cid} "
            f"channel={document.get('channel')!r} "
            f"status={document.get('status')!r} "
            f"type={document.get('type')!r} "
            f"title={document.get('title')!r} "
            f"content_type={document.get('content_type')!r} "
            f"content_url={document.get('content_url')!r} "
            f"patient_id={(patient or {}).get('id')!r} "
            f"available_document_types_count={len(available_types)}"
        )
        log.info(
            f"[doc_received] full cid={cid} "
            f"payload={json.dumps(ctx, default=str)}"
        )
        return []

You'll see two log lines per event in your worker syslog. The full event payload looks like:

{
  "document": {
    "id": "<task-uuid>",
    "channel": "fax",
    "content_url": "<file URL>",
    "content_type": "application/pdf",
    "title": "...",
    "type": "...",
    "status": "UNR",
    "created_at": "..."
  },
  "patient": {"id": "<key>"} | null,
  "available_document_types": [...]
}

Branch on document["channel"] == "fax" if you only care about faxes, then requests.get(document["content_url"]) to pull the file. There's a full lifecycle event family beyond just received — DOCUMENT_LINKED_TO_PATIENT, DOCUMENT_CATEGORIZED, DOCUMENT_REVIEWER_ASSIGNED, DOCUMENT_REVIEWED, DOCUMENT_DELETED — useful if you want to react at later steps too. Using the task UUID as cid lets you trace one document across all of them with one grep.

For processing documents that have already been triaged and attached to a clinical record — use the DocumentReference data class instead. It exposes document_url (a presigned S3 URL). This can be used once the document has been linked to the patient.

The polling case — "give me all unread integration tasks and let me fetch their files in a loop" — isn't supported via the SDK data class today (the file URL isn't projected onto IntegrationTask). But event-driven via DOCUMENT_RECEIVED covers the typical inbound-automation use case cleanly, so unless you specifically need polling, you should be set. You can always listen to this and save the document and the document ID, then use the IntegrationTask data module to update the status of whether the document is still unread.

But I did add the Github Issue #1663 to add the document to the IntegrationTask SDK module permanently, instead of relying on listening to the event because it would be nice to just have the document in the data module always. So I'll pass that back to our engineering team.