DocuLink Studio — Python SDK

Official Python SDK for the DocuLink Studio Customer API — OCR / LLM document processing and AI chat. It implements the full 59-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

🆕 What's new in 0.5.0

All-Process — hand the whole pipeline to the server instead of driving it yourself: pass AutoProcess to the upload, or call allProcess on an uploaded scan (it also resumes one stuck at "3"/"5"). Then poll getScanStatus until Done. scanDocument now defaults to this path and uses no Socket.IO at all, so it works from serverless functions, batch jobs and anything behind a firewall — pass allProcess: false for the 0.4.x socket-driven behaviour.

Human-in-the-loop — editOutputRecord / deleteOutputRecord change one row instead of PUT-ing the whole document back; getMappingCandidates / getMappingMeta / pickMappingCandidate let you inspect and change the master-data row the engine matched; exportScanOutput and getScanSourceFile download the result and the original file. listScans finally lists scans across every task.

⚠️ updateJsonOutput saves a full-output override that getJsonOutput returns in preference to the server's own output, so while one exists the record editors answer 409 on purpose — call clearOutputOverride first.

Row-level master data — read/add/update/delete individual rows plus search, history and a validating import, instead of replacing the whole table.

Account self-service — getMe, usage reports + export, invoices, rotateApiKey and getAuditLogs. Rotating the key revokes every previously issued token immediately; the SDK swaps in the new key and re-authenticates for you, but the key is shown only once — persist it.

⚠️ Breaking change in v0.4.0 — document status constants. DocumentStatus now matches the server's full lifecycle: "1" Uploaded, "2"/"3" OCR (processing/completed), "4"/"5" Schema, "6"/"7" Mapping, "9" Error — DocumentStatus.COMPLETED changed value from "4" to "7", and OCR / SCHEMA / MAPPING were replaced by *_PROCESSING / *_COMPLETED. See ../CHANGELOG.md.

⚠️ Also in v0.4.0: the doc-scan join payload is now the object {"room": ...} the server actually reads (bare-string joins were silently ignored — realtime never worked through 0.2.0–0.3.1); return_format_type takes "1"/"2" (the return format's Type), not "JSON"|"XML"|"CSV"; new scan_document pipeline helper + update_json_output.

⚠️ 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.

ℹ️ v0.3.1 — subscribe_document_scan(...) now sends the auth={"token": ...} Socket.IO handshake the document server requires. Authenticate (or pre-set an access token) before subscribing; anonymous connects are rejected.

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="you@example.com",          # 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

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

import uuid

task_id = str(uuid.uuid4())          # TaskId = UUID v4 ที่คุณสร้างเอง (**v4 เท่านั้น** — เวอร์ชันอื่นถูกปฏิเสธ)

result = client.upload_file(
    task_id=task_id,
    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="1",          # "1" public / "2" customer — ค่า Type ของ return format ที่เลือก (ไม่ใช่ชื่อ file format)
    client_uuid=None,                # optional
    filename="invoice.pdf",
)
doc_id = result.DocumentScanUUID

# Drive the pipeline (async — wait for each stage's odd status via realtime):
client.ocr_process(task_id, doc_id)      # → wait for status "3"
client.schema_process(task_id, doc_id)   # → wait for status "5"
client.mapping_process(task_id, doc_id)  # → wait for status "7"

output = client.get_json_output(task_id, doc_id)
print(output.JsonOutput)

Document status constants (even = stage running, odd = stage done — wait for the odd status of each stage before calling the next endpoint): DocumentStatus.UPLOADED ("1"), OCR_PROCESSING ("2"), OCR_COMPLETED ("3"), SCHEMA_PROCESSING ("4"), SCHEMA_COMPLETED ("5"), MAPPING_PROCESSING ("6"), COMPLETED ("7"), 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 object {"room": "doc-scan-<TaskUUID>"} (same shape as chat) 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.