omniscient-sdk

Official Python SDK for the Omniscient platform.

The SDK lets external services talk to an Omniscient backend over HTTP. A single class — Omniscient — wraps every endpoint the backend publishes under /sdk/*:

Capability	Token permission	What it does
External connectors	connector	Register a connector, run sync cycles, push files, query checksums.
Artifacts	mcp	Publish generated files for delivery in chat.
OCR	ocr	Extract Markdown/JSON/PDF from a document.
Data extraction	data-extraction	Extract structured data from a document against a saved template or inline schema.
Notifications	notification	Report a job's outcome (completed / failed); the platform e-mails its administrators.
Chat	chat	OpenAI-style chat completions (not implemented yet).

A token's permissions decide which methods are callable; the backend rejects calls made with a token that lacks the required permission. The chat methods raise NotImplementedError until the backend implementation lands.

Installation

pip install omniscient-sdk

Imports use the top-level omniscient package:

from omniscient import Omniscient

Authentication

All calls authenticate with an API token sent through the X-Api-Key HTTP header. Tokens are created from the Omniscient admin panel and carry one or more permissions (connector, mcp, ocr, chat).

The client reads two environment variables by default (or accepts them as constructor arguments):

Variable	Purpose
OMNISCIENT_API_URL	Base URL of the Omniscient backend.
OMNISCIENT_API_TOKEN	API token (omni_...).

from omniscient import Omniscient

# Reads OMNISCIENT_API_URL / OMNISCIENT_API_TOKEN from the environment.
with Omniscient() as client:
    print(client.connector_id)        # connector bound to this token, if any
    print(client.last_sync_started_at)  # checkpoint of the last sync run

# …or pass them explicitly:
client = Omniscient(api_url="https://omniscient.example.com", api_token="omni_…")

When the token is bound to a connector, the client resolves the connector ID on construction. Pass resolve_connector=False to skip that lookup when you only use the MCP or OCR features.

External connectors

A connector pushes documents from an external source into Omniscient. The run_sync helper drives a full cycle — optional auto-register → notify start → your sync function → notify end (with stale-item cleanup):

import os
from pathlib import Path

from omniscient import Omniscient


def sync(client: Omniscient) -> list[str]:
    source_dir = Path(os.environ["OMNISCIENT_SOURCE_DIR"])
    existing = set(client.get_existing_checksums())
    active: list[str] = []

    for path in sorted(source_dir.rglob("*")):
        if not path.is_file():
            continue
        checksum = Omniscient.compute_checksum(path)
        active.append(checksum)
        if checksum not in existing:
            client.push_file(path, source_id=str(path.relative_to(source_dir)))

    # Returning the active checksums lets the backend delete stale items.
    return active


if __name__ == "__main__":
    with Omniscient() as client:
        client.run_sync(sync, name="local-files", description="Local files")

item_exists(source_id=...) / item_exists(checksum=...) query the backend for a single item without fetching the whole checksum list. See examples/external_connector/ for a runnable connector and Dockerfile.

Artifacts

Publish a file produced by an MCP tool as an Artifact, and fetch it back by id:

from omniscient import Omniscient

with Omniscient(resolve_connector=False) as client:
    info = client.artifact_upload("./report.pdf", display_name="Q4 report")
    client.artifact_download(info["artifactId"], "./downloaded.pdf")

Uploading alone only stages the file. To deliver it to the chat user, the MCP tool result must declare the id under the reserved omniscient_artifacts key of its structured content:

{
  "omniscient_artifacts": [
    {"artifact_id": "<artifactId>", "filename": "report.pdf",
     "content_type": "application/pdf", "size": 12345}
  ]
}

Only artifact_id is required — the other fields are hints. The chat backend then shows the file as a downloadable card on the assistant's message and in the user's Artifacts list. Identical re-uploads by the same token are deduplicated server-side (deduped: true in the response).

Document inputs handed to an MCP tool arrive as attachment keys of the form "<attachment_id>/<filename>"; fetch them with:

client.attachment_download("<attachment_id>/report.docx", "./input.docx")

The deprecated mcp_upload_attachment / mcp_download_attachment aliases (old /sdk/mcp/* routes) were removed in 0.5.0 — use artifact_upload / artifact_download.

MCP helpers

When the Omniscient platform invokes a first-party MCP server, it forwards a per-request API token in the X-Omniscient-Token HTTP header, so every Omniscient call the tool makes happens on behalf of the calling organization. omniscient.mcp_helpers turns the current request's headers into a configured client — framework-agnostic, no MCP imports:

from omniscient.mcp_helpers import client_from_headers

def my_tool(...):
    headers = ...  # whatever your MCP framework exposes for the request
    client = client_from_headers(headers, api_url="https://<backend>")
    if client is None:
        raise RuntimeError("no Omniscient credentials for this request")
    with client:
        info = client.artifact_upload("./report.xlsx")

Resolution order: the X-Omniscient-Token header (case-insensitive) first, then — unless fallback_env=False — the server-wide OMNISCIENT_API_TOKEN environment variable, else None. The client is built with resolve_connector=False and should be closed after the request.

OCR

Run OCR on a single document. By default the structured result is returned as a dict; pass output_format together with dest to download the rendered file instead:

from omniscient import Omniscient

with Omniscient(resolve_connector=False) as client:
    result = client.ocr_extract("./document.pdf", mode="STRUCTURED")
    print(result["markdown"])

    # Render and download a file:
    client.ocr_extract(
        "./document.pdf", output_format="MARKDOWN", dest="./document.md"
    )

mode accepts "PLAIN", "STRUCTURED" (default) or "VLM"; output_format accepts "MARKDOWN", "JSON" or "PDF". Result keys are snake_case (markdown, regions, page_count, …).

Data extraction

Extract structured data from a document against a saved template or an inline definition (provide exactly one). Non-PDF inputs are converted to PDF server-side. The structured result is returned as a dict; pass output_format ("csv"/"xlsx") together with dest to download a table:

from omniscient import Omniscient

with Omniscient(resolve_connector=False) as client:
    # Against a saved template:
    result = client.extraction_extract("./invoice.pdf", template_id="<id>")
    for field in result["fields"]:
        print(field["field_path"], field["value"], field["confidence"])

    # Against an inline definition + a hint:
    client.extraction_extract(
        "./invoice.pdf",
        definition={"fields": [{"key": "total", "label": "Total", "type": "currency"}]},
        hints="the grand total is bottom-right",
    )

    # Download a CSV of the extracted fields:
    client.extraction_extract(
        "./invoice.pdf", template_id="<id>", output_format="csv", dest="./out.csv"
    )

Result keys are snake_case: data (the structured record), fields (each with field_path, value, value_type, confidence, source_spans) and usage.

Notifications

Report the outcome of an unattended job — INFO when it completed, WARNING when it failed — and let the platform deliver an e-mail to the recipients its administrators configured. You only provide the content: the subject, a plain-text message and optional key/value details rendered as a table. The e-mail template (English) and the recipient list are owned by the platform; the token's name appears as the source.

from omniscient import Omniscient

with Omniscient(resolve_connector=False) as client:
    record = client.notify_warning(
        "BigQuery synchronization failed",
        "The scheduled sync did not complete.",
        details={"Job": "ds_sync", "Error": "TimeoutError: relay unreachable"},
    )
    print(record["id"], record["status"])  # e.g. "PENDING"

Delivery is asynchronous: the call returns the notification record with status == "PENDING" (or "SKIPPED" when successful completions are silenced on the platform). Pass wait=True to poll until it becomes SENT or FAILED, or check later with client.get_notification(record["id"]).

For cron-style jobs, job_notifier() sends exactly one notification per run and never raises from the notification itself:

with Omniscient(resolve_connector=False) as client:
    with client.job_notifier("BigQuery synchronization") as run:
        uploaded = sync_tables()                 # raises → WARNING "… failed"
        run.details["Rows uploaded"] = f"{uploaded:,}"
    # normal exit → INFO "BigQuery synchronization completed" (+ Duration)

Errors surface as httpx.HTTPStatusError: 403 when the token lacks the notification permission, 429 when a safety limit is hit (see the Retry-After header), 503 when notifications are disabled on the platform. Tokens that only hold the notification permission can be used with the default constructor: the connector lookup silently yields connector_id=None.

Assistants — User Guide documents

Each assistant can carry one User Guide (PDF or DOCX) that chat users open from the assistant's info and from the bottom of its starter message. A token with the assistant permission can list the assistants and upload, replace or remove that document — typically from a cronjob that keeps the manuals in sync with files produced elsewhere:

from omniscient import Omniscient

with Omniscient(resolve_connector=False) as client:
    assistant = client.assistant_find("Legal Buddy")  # by its unique name
    summary = client.assistant_user_guide_upload(
        assistant["id"], "/guides/legal-buddy.pdf"  # PDF or DOCX
    )
    print(summary["userGuide"]["checksum"])

assistant_user_guide_upload() compares the local SHA-256 with the remote userGuide.checksum and returns the current summary without uploading when they match (pass skip_unchanged=False to force the upload). The server validates the bytes, not the declared type: 422 means the file is not a PDF/DOCX matching its extension, 413 that it exceeds the platform's size cap. assistant_user_guide_delete() removes the document (idempotent). assistant_list() / assistant_get() return the summaries (id, name, description, isVisible, lastUpdate, userGuide).

See examples/user_guide_sync/ for a cron-ready script.

Chat (not implemented)

from omniscient import Omniscient

with Omniscient(resolve_connector=False) as client:
    # Raises NotImplementedError today.
    client.chat_completions({"messages": [{"role": "user", "content": "Hi"}]})

Development

The project uses uv and ruff (pinned in pyproject.toml).

uv sync
uv run ruff check src tests
uv run ty check
uv run pytest

Versioning & release

The package version lives in [project].version of pyproject.toml (a single source of truth; omniscient.__version__ is read from the installed package metadata). Releases are driven by CI:

• pushes to develop publish a dev build to TestPyPI (the version is suffixed with .devN so each build is unique);
• pushes to main publish the exact pyproject.toml version to PyPI as omniscient-sdk.

Bump version in pyproject.toml before promoting a release to main.