clark-platform-client

Typed, standalone Python client (sync + async) for the Clark Platform API — the OpenAI-compatible-ish public HTTP API for creating Clark agent runs, streaming their progress, and reading back chat-completion/response objects, usage, artifacts, and durable memory. It also covers Clark-owned file uploads and repository commit context for agent integrations.

This client implements the wire contract documented in platform/clients/API_CONTRACT.md in the Clark monorepo. If something here disagrees with that file, the contract file is the source of truth.

• Only required dependency: httpx.
• No pydantic — models are plain stdlib dataclasses.
• Both a sync client (ClarkClient) and an async client (AsyncClarkClient); pick one without pulling in asyncio machinery you don't need.
• Python 3.9+.

Install

Published on PyPI:

uv add clark-platform-client
# or: pip install clark-platform-client

Authentication

Platform API keys (ck_live_...) are minted from the signed-in /platform page in the Clark web app — there is no public self-serve key-creation endpoint. Every key carries a fixed scope set (responses:create, responses:read, models:read, artifacts:read, memories:read, files:read, files:write).

from clark_platform import ClarkClient

client = ClarkClient(api_key="ck_live_xxxxxxxxxxxxxxxxxxxx")

By default the client talks to production (https://www.clarkchat.com).

Quickstart — sync

from clark_platform import ClarkClient

with ClarkClient(api_key="ck_live_...") as client:
    models = client.models.list()
    print([m.id for m in models.data])

    response = client.responses.create(model="clark", input="Summarize this repo's README.")
    print(response.status, response.output_text)

    chat = client.chat.completions.create(
        model="openrouter:qwen36_flash",
        messages=[{"role": "user", "content": "hello"}],
    )
    print(chat.output_text)

Quickstart — async

import asyncio
from clark_platform import AsyncClarkClient


async def main() -> None:
    async with AsyncClarkClient(api_key="ck_live_...") as client:
        response = await client.responses.create(model="clark", input="hello")
        print(response.output_text)


asyncio.run(main())

Two execution modes: agentic vs. provider gateway

The API has two entirely different behaviors depending on model:

1. Agentic tiers (clark, clark_max, openrouter:*) — Clark runs its own internal agent loop and returns only the final projected answer. tools/tool_choice are not accepted by these methods (the server rejects them with 400 unsupported_parameter, so the SDK doesn't even expose those parameters on responses.create/chat.completions.create).
2. Provider-qualified author/model gateway IDs — your messages (including prior tool_calls/tool messages) are forwarded verbatim to the underlying OpenRouter-compatible model, and native OpenAI-format tool_calls come back for you to execute yourself. Clark does not run tools for this tier. clark-code remains a compatibility alias. This Python client's typed passthrough helper uses /v1/chat/completions; the REST API also exposes provider-qualified passthrough on /v1/responses.

Because these are different response shapes, the passthrough tier is exposed via clearly separate methods that return the raw upstream JSON dict instead of a typed Clark object:

import os

# Agentic — typed ChatCompletionObject, no tools allowed.
chat = client.chat.completions.create(model="clark", messages=[...])

# Passthrough — raw OpenAI-compatible dict, tools/tool_choice forwarded verbatim.
raw = client.chat.completions.create_passthrough(
    model=os.environ["CLARK_MODEL"],
    messages=[...],
    tools=[{"type": "function", "function": {...}}],
)

Streaming

responses.stream(...)

Named SSE events (response.created, response.output_text.delta, response.artifact.completed, response.usage.updated, response.completed/response.failed, ...). Per the contract, the full answer arrives as a single response.output_text.delta event, not token-by-token — the iterator API is for symmetry with chat-completions streaming and future finer-grained deltas.

for event in client.responses.stream(model="clark", input="hello"):
    if event.type == "response.output_text.delta":
        print(event.delta, end="")
    elif event.type == "response.completed":
        print("\n--- done ---", event.response.status)

Async:

async for event in async_client.responses.stream(model="clark", input="hello"):
    ...

chat.completions.stream(...)

Plain data:-only SSE (no event: name), chat.completion.chunk objects, terminated by the server's literal data: [DONE] (already consumed for you — it never appears as a yielded item):

for chunk in client.chat.completions.stream(
    model="clark",
    messages=[{"role": "user", "content": "hello"}],
    stream_options={"include_usage": True},
):
    for choice in chunk.choices:
        if choice.delta.content:
            print(choice.delta.content, end="")
    if chunk.usage:
        print("\nusage:", chunk.usage)

chat.completions.stream_passthrough(...)

Same framing, but yields raw upstream JSON dicts (native tool_calls deltas and all) instead of a typed ChatCompletionChunk, since the passthrough tier's chunk shape is whatever the upstream OpenAI-compatible provider sends.

Other endpoints

# Poll a response (e.g. after background=True, or to recover from a chat
# completions timeout by replacing the "chatcmpl_" prefix with "resp_").
response = client.responses.get("resp_01jz4n8h2f7g9k4q2m6s")

# Progress events for a response.
events = client.responses.list_events("resp_01jz4n8h2f7g9k4q2m6s", after_seq=0, limit=200)

# Durable memory.
memories = client.memories.list(q="deploy checklist")

# Upload, list, retrieve, and delete caller-owned files. Use the returned
# clark_file_* id in later gateway requests; upstream ids are never exposed.
uploaded = client.files.create(
    b"release notes",
    filename="notes.txt",
    mime_type="text/plain",
    purpose="assistants",
)
files = client.files.list(limit=20)
metadata = client.files.retrieve(uploaded.id)
deleted = client.files.delete(uploaded.id)

# Bounded commit evidence for a repository previously indexed by Clark Desktop.
context = client.code.repositories.context(
    "repository-fingerprint",
    q="authentication changes",
    limit=8,
)

# Membership-checked organizational knowledge available to the caller.
knowledge = client.organization_knowledge.search(
    query="authentication changes",
    limit=8,
)

# Server-owned image generation accepts only base64 data-image references.
image = client.images.generate(
    prompt="Turn this into a warm editorial illustration.",
    input_images=["data:image/png;base64,iVBORw0KGgo..."],
    idempotency_key="image-retry-1",
)

# Raw HTTP response for range-capable artifact downloads.
artifact = client.artifacts.download(
    "conversation-id", "report.pdf", download=True, headers={"Range": "bytes=0-1023"}
)

The async client exposes the same methods under async_client.files and async_client.code.repositories; await each call.

Errors

All non-2xx responses raise ClarkApiError, parsed from the API's error envelope:

from clark_platform import ClarkApiError

try:
    client.responses.create(model="clark", input="hello")
except ClarkApiError as err:
    print(err.status_code, err.type, err.code, err.param, err.message)

Network-level failures (connection errors, timeouts, DNS failures) are not wrapped — they surface as the underlying httpx exception (e.g. httpx.ConnectError, httpx.TimeoutException), so you can rely on httpx's own exception hierarchy for those.

Timeouts

Non-background responses/chat.completions calls can legitimately take up to ~120s server-side (CLARK_PLATFORM_RESPONSE_WAIT_MS, default 120000ms). The client defaults its httpx timeout to 150s to comfortably clear that; pass your own timeout= (or a pre-configured http_client=) to ClarkClient/AsyncClarkClient to change it.

Development

cd platform/clients/python
uv sync --python 3.12
uv run --python 3.12 pytest -q

examples/live_smoke.py is an opt-in, real-network smoke test — it reads CLARK_API_BASE_URL, CLARK_API_KEY, and CLARK_TEST_MODEL from the environment and exits early with a clear message if CLARK_API_KEY is unset. It is not part of the test suite and is never run automatically.