callosum-sdk

callosum-sdk provides an AsyncOpenAI-compatible client for latency-insensitive models running on spot capacity. A familiar chat-completions call becomes a background submit/poll operation without holding an HTTP connection across a GPU cold start.

Install

pip install callosum-sdk

Quickstart

import asyncio
import os

from callosum_sdk import AsyncOpenAI


async def main() -> None:
    async with AsyncOpenAI(
        api_key=os.environ["CALLOSUM_API_KEY"],
        base_url=os.environ.get("CALLOSUM_BASE_URL", "https://api.callosum.ai/v1"),
    ) as client:
        response = await client.chat.completions.create(
            model="qwen3.5-4b",
            messages=[{"role": "user", "content": "What is the corpus callosum?"}],
        )
        print(response.choices[0].message.content)


asyncio.run(main())

Run uv run demo-readme from a source checkout to execute that exact Quickstart block. Set CALLOSUM_BASE_URL to target a staging gateway or port-forward; otherwise it defaults to the public API.

The SDK sends the public slug as spot/qwen3.5-4b, which selects the gateway's spot proxy. The gateway maps that slug through its deployed model catalogue and forwards to the spot orchestrator's background API. The returned response still reports qwen3.5-4b, keeping the routing namespace transparent to application code.

Available models

Pass the slug alone and it is served on spot capacity; the SDK adds the namespace.

The slug set is the gateway's, not the SDK's: a syntactically valid slug with no deployed serving profile returns unknown_model, so a new model becomes available without an SDK release.

Passthrough models

Some models are not ours to schedule: the gateway proxies them to an always-on upstream and returns the reply on the same connection. Name one by its namespace and the SDK sends it straight through, so the choice is a model name rather than a different client.

response = await client.chat.completions.create(
    model="cerebras/gpt-oss-120b",
    messages=[{"role": "user", "content": "What is the corpus callosum?"}],
)

The rule is the namespace: a slug with no / is spot-backed and goes through the background submit/poll protocol, while a slug carrying one names a passthrough upstream and goes to the gateway's ordinary chat completions route. That is the same line the gateway itself draws, so the two cannot disagree.

Because a passthrough answers immediately, it keeps what the background protocol has to give up — stream=True works:

stream = await client.chat.completions.create(
    model="cerebras/gpt-oss-120b",
    messages=[{"role": "user", "content": "Explain the corpus callosum."}],
    stream=True,
)
async for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="")

Fixed workflows

Use a fixed workflow without exposing its metaprogrammer DAG or internal input shape:

from callosum_sdk import OpenAI

with OpenAI(
    api_key=os.environ["CALLOSUM_API_KEY"],
    base_url=os.environ.get("CALLOSUM_BASE_URL", "https://api.callosum.ai/v1"),
) as client:
    response = client.workflows.create(
        workflow="autoheal",
        payload={"source": "https://github.com/callosumtechnologies/some-repo"},
    )

    schema_snapshot = {"tables": [{"name": "agencies"}]}
    response = client.workflows.create(
        workflow="database-edit",
        payload={
            "request": "Make phone unique on agencies.",
            "schema_snapshot": schema_snapshot,
            "database_dsn": "postgresql://readonly:...@db.example/postgres",
        },
    )

Use AsyncOpenAI and await client.workflows.create(...) for the same request without blocking an async application.

See fixed workflows for the translated gateway request and fixed defaults.

deepseek-v4-flash is a reasoning model that answers directly unless asked to deliberate. Request deliberation per call:

response = await client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user", "content": "What is the corpus callosum?"}],
    extra_body={"chat_template_kwargs": {"thinking": True}},
)
print(response.choices[0].message.reasoning_content)
print(response.choices[0].message.content)

The deliberation arrives as reasoning_content on the message, leaving content as the answer alone. It is an additional field rather than part of the typed OpenAI surface, so it is readable but invisible to type checkers.

Behaviour and limits

• chat.completions.create() is non-streaming: it submits immediately, polls every five seconds, and returns a typed OpenAI ChatCompletion at completion.
• The default one-hour client wait can be changed with request_timeout_seconds, and a request declaring deadline_seconds raises it to outlive that deadline; the request continues in the orchestrator if a local caller is cancelled.
• Polling waits out a briefly absent endpoint rather than failing the call, so a deploy of the spot orchestrator costs a pause instead of a lost response. See the spot client guide.
• Polls remain authenticated and tenant-scoped, but do not consume the customer's request-per-minute API limit.
• Every submit carries an Idempotency-Key, so a submit whose response is lost is retried onto the request it already opened rather than running the completion twice. See the spot client guide.
• model_prefix="spot/" is idempotent. Passing an already namespaced model does not produce spot/spot/.... Setting it to another background namespace points the client at a different endpoint; switch/ reaches the always-on small-model fleet, described in the switch guide.
• The API key identifies the tenant. A response id submitted by one tenant cannot be polled by another.
• Raw Files and Batch API batching remains available through the explicit BatchOpenAI class for non-spot workload batches.

See the spot client guide for the routing contract, the switch guide for the always-on small-model fleet, and the batching guide for BatchOpenAI.