doculink-studio 0.3.0

Official Python SDK for the DocuLink Studio Customer API (OCR / LLM document processing + AI chat).

DocuLink Studio — Python SDK

Official Python SDK for the DocuLink Studio Customer API — OCR / LLM document processing and AI chat. It implements the full 32-endpoint contract with automatic authentication, token auto-refresh, typed models, multipart uploads, SSE chat streaming and Socket.IO real-time subscriptions.

• Distribution name (PyPI): doculink-studio
• Import package: doculink_studio

⚠️ Breaking change in v0.3.0 — real-time subscription. subscribe_document_scan(...) now takes the TaskUUID (the task id you use for upload), not the DocumentScanUUID. The server broadcasts doc-scan events to room doc-scan-{TaskUUID}; a 0.2.0 subscription that passed the scan UUID received nothing. Each event's payload UUID is still the DocumentScanUUID. See ../CHANGELOG.md.

Install

pip install doculink-studio

Requires Python 3.9+. Runtime dependencies: requests, pydantic>=2, python-socketio[client], websocket-client.

Quickstart

from doculink_studio import DoculinkClient

client = DoculinkClient(
    provider_api_key="<35-char provider key>",
    customer_api_key="<35-char customer key>",
    email="[e-mail]",          # optional
    # base_url / document_ws_url / chat_ws_url default to the test environment
)

# Authentication is automatic on the first call, but you can force it:
client.authenticate()

usage = client.get_usage()
print(usage.planCode, usage.quotaRemaining)

Configuration

Kwarg	Default
base_url	https://test-api-provider.doculink.studio/api/v1
document_ws_url	https://test-ws.doculink.studio
chat_ws_url	https://test-ws.doculink.studio:8000
timeout	30 (SSE / sync chat use 300s automatically)
access_token / refresh_token	optional — pre-set to skip initial auth
session	optional requests.Session

Authentication & auto-refresh

The client stores the access + refresh tokens after authenticate(). Every authenticated request sets Authorization: Bearer <AccessToken>. On a 401 the client transparently refreshes the access token and retries once; if refresh fails it re-authenticates with the API keys and retries once.

Upload & process a document

result = client.upload_file(
    task_id="my-task",
    file=b"...pdf bytes...",         # bytes, a file path (str), or a stream
    schema_uuid="<schema uuid>",
    return_format_uuid="<return format uuid>",
    return_format_type="JSON",       # JSON | XML | CSV
    client_uuid=None,                # optional
    filename="invoice.pdf",
)
doc_id = result.DocumentScanUUID

# Drive the pipeline (each step returns a status string):
client.ocr_process("my-task", doc_id)
client.schema_process("my-task", doc_id)
client.mapping_process("my-task", doc_id)

output = client.get_json_output("my-task", doc_id)
print(output.JsonOutput)

Document status constants: DocumentStatus.OCR ("1"), SCHEMA ("2"), MAPPING ("3"), COMPLETED ("4"), ERROR ("9").

Real-time — document processing (Socket.IO)

from doculink_studio import DocScanUpdate

def on_update(u: DocScanUpdate):
    print(u.Status, u.CurrentLog)

sub = client.subscribe_document_scan(task_id, on_update=on_update)
# ... later ...
sub.close()

The subscribe id is the TaskUUID — the same :taskid you upload to — not the DocumentScanUUID. On connect the SDK emits join with the plain string "doc-scan-<TaskUUID>" and dispatches doc-scan events parsed into DocScanUpdate. A task may hold several scans through the one room; each update's UUID field is the DocumentScanUUID it belongs to.

Chat

session_id = client.create_chat_session(model="gpt-4o-mini")

# Synchronous (blocks until the full reply is ready):
msg = client.send_chat_message_sync(session_id, "Summarise the invoice")
print(msg.Content)

# Streaming over SSE:
def on_event(evt):
    if "chunk" in evt:
        print(evt["chunk"], end="")

final = client.send_chat_message_stream(session_id, "Explain more", on_event)
print("\nFinal:", final.Content)

# History / listing
sessions = client.list_chat_sessions(status="active", page=1, limit=20)
history = client.get_chat_history(session_id, page=1, limit=50)

Chat over Socket.IO

sub = client.subscribe_chat_session(
    session_id,
    on_start=lambda e: print("start", e),
    on_chunk=lambda chunk, e: print(chunk, end=""),
    on_end=lambda msg, e: print("\ndone:", msg.Content),
    on_error=lambda e: print("error", e),
)
client.send_chat_message(session_id, "hello")   # async (202) — watch the socket
# ...
sub.close()

The chat join emits an object {"room": "chat-<sessionID>"} and passes auth={"token": <access token>} on the handshake when a token is available.

Error handling

Every method raises ApiError on HTTP >= 400 or an envelope status: false (including billing rejections that come back as HTTP 200 with status: false).

from doculink_studio import ApiError

try:
    client.get_usage()
except ApiError as e:
    print(e.status_code, e.message, e.status)

Development

python -m venv .venv
source .venv/Scripts/activate     # Git Bash on Windows
pip install -e ".[dev]"
pytest -q

More docs

See the bundled API reference at ../docs/index.html and the canonical contract in ../CONTRACT.md.