clark-platform-client 0.1.0

Typed sync/async Python client for the Clark Platform API

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.

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

Not published to PyPI yet. Install directly from a local checkout of this directory:

# with uv, inside another project
uv add /path/to/clark/platform/clients/python

# or with pip, editable install for local development
pip install -e /path/to/clark/platform/clients/python

Authentication

Platform API keys (clk_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).

from clark_platform import ClarkClient

client = ClarkClient(api_key="clk_live_xxxxxxxxxxxxxxxxxxxx")

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

Quickstart — sync

from clark_platform import ClarkClient

with ClarkClient(api_key="clk_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:qwen35_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="clk_live_...") as client:
        response = await client.responses.create(model="clark", input="hello")
        print(response.output_text)


asyncio.run(main())

Two families of tier: agentic vs. clark-code passthrough

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. The clark-code passthrough tier — 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. This is only available via /v1/chat/completions, never /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:

# 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="clark-code",
    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")

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.