Continuum Task Server SDK for Python

Python client for the Continuum task server. Designed so a 20-line script can stand up a worker that claims queue items, runs your code, and reports results.

• TaskServer — decorator-based worker loop: claim, heartbeat while the handler runs, status updates, backoff, graceful shutdown. Long-running claims (auto_complete=False) need your heartbeat loop (documented below).
• ContinuumClient — thin pythonic client over the management + queue REST APIs (task types, task items, versions, queue, content store).

Requires Python 3.10+.

Installation

Tagged releases (v*) are published to PyPI:

pip install continuum-task-server-sdk

Dev and PR builds are still attached to private GitHub Releases. Install those with a token:

pip install \
  "https://${GITHUB_TOKEN}@github.com/ContinuumWorkflow/continuum-task-server-sdk-python/releases/download/v0.1.0/continuum_task_server_sdk-0.1.0-py3-none-any.whl"

GITHUB_TOKEN must be a PAT with repo read access.

Quickstart: build a task server in 20 lines

import os
from continuum_task_server import TaskServer

server = TaskServer(
    base_url=os.environ["CONTINUUM_URL"],
    api_key=os.environ["CONTINUUM_API_KEY"],
)

@server.task("echo")
def echo(item):
    return {"echoed": item.input_data_json}

@server.task("greet")
def greet(item):
    name = (item.input_data_json or {}).get("name", "world")
    return {"message": f"hello, {name}"}

if __name__ == "__main__":
    server.run()

Run it. The server polls /api/queue/claim for echo and greet, claims items as they become available, runs your handler in a thread, heartbeats while it runs, and:

• Handler returns a value → queue item marked ENDED, return value JSON-encoded as outputData (default auto_complete=True).
• Handler raises → queue item marked KILLED, error info written to outputData.

Press Ctrl+C (or send SIGTERM) and the server stops polling and waits up to shutdown_timeout seconds for in-flight handlers to finish.

Deferred completion (auto_complete=False)

The handler runs under the normal in-handler heartbeat (same as any task). When it returns, the server does not send ENDED and does not keep heartbeating: you own the claim until you finish or lose it to timeouts.

Typical pattern: persist item.id (and anything else you need), then on each pass of your poller (cron, loop, worker restart), call server.client.queue.heartbeat(queue_item_id) so the Continuum claim stays alive, and when the real-world condition is met call server.complete_queue_item(...) or server.fail_queue_item(...). Use the same worker API key as the process that claimed the item (often the same TaskServer / ContinuumClient config loaded from env).

@server.task("wait-for-mail", auto_complete=False)
def wait_for_mail(item):
    db.insert_outstanding(queue_item_id=str(item.id), payload=item.input_data_json)
    # Returns without ENDED — no background heartbeat from TaskServer

# Elsewhere: each time you poll your DB for outstanding work (including after restart):
for row in db.outstanding_rows():
    server.client.queue.heartbeat(row.queue_item_id)
    if mail_arrived(row):
        server.complete_queue_item(row.queue_item_id, output_data={"received": True})

Standalone process (no TaskServer): build a ContinuumClient with the worker key and call client.queue.heartbeat / client.queue.update_status the same way.

TaskServer options

TaskServer(
    base_url="http://localhost:8080",
    api_key="...",
    max_workers=4,            # thread pool size across all tasks
    poll_interval=1.0,        # initial poll delay (backs off when idle)
    max_poll_interval=5.0,    # max idle poll delay
    heartbeat_interval=15.0,  # how often to call /heartbeat per running task
    shutdown_timeout=30.0,    # how long to wait for handlers during shutdown
)

Per-task concurrency limit:

@server.task("docker-run", concurrency=2)
def run_docker(item):
    ...

Handler signature: def handler(item: QueueItem) -> dict | list | str | None. Inside, you have:

• item.input_data — raw JSON string from the queue item (or None).
• item.input_data_json — parsed value (dict / list / str / None).
• server.client — full ContinuumClient if you need to chain management calls, fetch content, enqueue child tasks, etc.

Using ContinuumClient directly

from continuum_task_server import ContinuumClient, TaskStatus

with ContinuumClient(base_url="http://localhost:8080", api_key="...") as client:
    # Task types
    types = client.task_types.list()
    echo_type = client.task_types.get_by_name("echo")

    # Task items + versions
    item = client.task_items.get_by_name("my-task")
    versions = client.task_items.versions.list(item.id)

    # Enqueue work
    queued = client.queue.add(task_name="echo", input_data={"hello": "world"})

    # Worker-side primitives (normally handled by TaskServer)
    claimed = client.queue.claim("echo")
    if claimed is not None:
        client.queue.heartbeat(claimed.id)
        client.queue.update_status(claimed.id, TaskStatus.ENDED, output_data={"ok": True})

    # Content store
    content = client.content_store.get_by_url("db://...")
    if content is not None:
        print(content.as_text())

input_data / output_data accept dict / list / str / None; non-string values are JSON-encoded for you.

Errors

All API failures raise ContinuumError or a specific subclass: BadRequestError, UnauthorizedError, ForbiddenError, NotFoundError, ConflictError, ServerError. Each carries status_code and body.

from continuum_task_server import ContinuumClient, NotFoundError

with ContinuumClient(...) as client:
    try:
        client.task_types.get_by_name("does-not-exist")
    except NotFoundError as e:
        print(e.status_code, e.body)

Endpoints covered

All requests send Api-Key: <your-key>. 204 No Content responses are normalized to None (e.g. client.queue.claim() returns None when nothing's available).

Development

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

ruff check .
ruff format --check .
pytest

Tests use respx to mock the httpx transport — no live server required.

Versioning

• Tag v1.2.3 → release wheel 1.2.3 attached to a v1.2.3 GitHub Release.
• Push to main → prerelease wheel 0.1.0.dev{run}+{shortsha} attached to a dev-{shortsha} Release.
• Pull request → prerelease wheel 0.1.0b{pr}.{run} attached to a pr-{pr} Release; install command posted as a PR comment.

License

MIT.