LatentKit Python SDK

Official Python client for the canonical LatentKit /v1 API.

Route-based by design: do not pass model, provider, route, or policy in SDK requests. The API key's assigned published route selects the provider/model at runtime.

Install

pip install latentkit

Requires Python 3.10+.

Quickstart

from latentkit import LatentKit, LatentKitAPIError

with LatentKit(api_key="YOUR_RAW_KEY") as client:
    try:
        response = client.chat.create(
            messages=[{"role": "user", "content": "Say hello from LatentKit."}],
            max_tokens=100,
            response_profile="balanced",
        )
        print(response["content"])
    except LatentKitAPIError as exc:
        print(exc.status_code, exc.code, exc.body)

Async

import asyncio

from latentkit import AsyncLatentKit


async def main() -> None:
    async with AsyncLatentKit(api_key="YOUR_RAW_KEY") as client:
        response = await client.completions.create(
            prompt="Write a short product description for LatentKit.",
            system="Respond in one sentence.",
        )
        print(response["content"])


asyncio.run(main())

Route-based requests

The SDK is not model-based. Your application sends the task body and optional response_profile; LatentKit resolves the provider/model from the API key's assigned route. model in a response is reporting metadata for the route that won, not a request field.

SDK calls reject route-control keys such as model, provider, route, and policy, including inside extra_body.

Inspect the connected route

client.me.retrieve() and await client.me.retrieve() are typed as MeResponse. The response includes connection, assigned route, ordered model, credit, and latest winning request data when available:

context = client.me.retrieve()
route = context.get("policy")

if route:
    print(route.get("name"), route.get("model_count"))
    for model in route.get("models", []):
        print(model.get("rank"), model.get("provider"), model.get("model"))

latest = context.get("latest_request")
if latest:
    print(latest.get("model"))

The latest winner is request activity, not a permanently selected model. Runtime SDK credentials cannot change the assigned route.

Client options

• api_key is required.
• base_url defaults to https://ai.latentkit.com and is normalized to /v1.
• timeout defaults to 120.0.
• headers lets you add extra request headers.
• http_client lets you inject a custom httpx.Client or httpx.AsyncClient.

LatentKit(...) and AsyncLatentKit(...) create httpx clients with a default timeout of 120s.

If you inject your own http_client, configure timeouts on that client yourself:

import httpx
from latentkit import LatentKit

http_client = httpx.Client(timeout=30.0)
client = LatentKit(api_key="YOUR_RAW_KEY", http_client=http_client)

Streaming

from latentkit import LatentKit

with LatentKit(api_key="YOUR_RAW_KEY") as client:
    for event in client.chat.stream(
        messages=[{"role": "user", "content": "Count from one to five."}],
    ):
        if event.event == "error":
            raise RuntimeError(event.data)
        if event.is_done:
            break
        print(event.data["delta"], end="")

Response profiles

Pass response_profile directly to ask the assigned policy for a speed/depth tradeoff:

from latentkit import LatentKit

with LatentKit(api_key="YOUR_RAW_KEY") as client:
    response = client.chat.create(
        messages=[{"role": "user", "content": "Give me the short version."}],
        response_profile="fast",
    )

Allowed values are fast, balanced, and deep. The assigned policy controls whether request overrides are allowed and which routes are eligible for each profile.

Chat, image, and embeddings

from latentkit import LatentKit

with LatentKit(api_key="YOUR_RAW_KEY") as client:
    chat = client.chat.create(
        messages=[{"role": "user", "content": "Write one sentence about LatentKit."}],
    )
    image = client.image.generate(prompt="A clean product icon", size="1024x1024")
    vectors = client.embeddings.create(input=["hello world"], dimensions=256)

Agent sessions

from latentkit import LatentKit

with LatentKit(api_key="YOUR_RAW_KEY") as client:
    session = client.agents.sessions.create(
        task="Inspect the repo and explain the auth flow",
        workspace_root="/workspace",
        permission_mode="workspace-write",
    )
    queued = client.agents.sessions.run(session["id"])
    print(queued)

See docs/latentkit-coder-api.md for the full agent session request/response model.

Modalities

with LatentKit(api_key="YOUR_RAW_KEY") as client:
    client.embeddings.create(input=["hello world"], dimensions=256)
    client.image.generate(prompt="A clean product icon", size="1024x1024")
    client.speech.create(input="Hello from LatentKit.", voice="alloy")
    client.transcription.create(audio={"base64": "..."}, language="en")
    client.translation.create(audio={"base64": "..."}, target_language="en")
    client.video.generate(prompt="A short product scene", duration_seconds=4)

Queue requests support every endpoint accepted by POST /v1/queue and an optional idempotency key:

client.queue.create(
    endpoint="embeddings",
    payload={"input": ["hello world"]},
    idempotency_key="import-row-42",
)

Queue support is enqueue-only. There is no public per-job result endpoint, so workflows that need results should use their host job runner and call a synchronous SDK resource from that job.

Supported resources

• client.chat.create(...)
• client.chat.stream(...)
• client.completions.create(...)
• client.completions.stream(...)
• client.vision.create(...)
• client.vision.stream(...)
• client.embeddings.create(...)
• client.image.generate(...)
• client.speech.create(...)
• client.transcription.create(...)
• client.translation.create(...)
• client.video.generate(...)
• client.queue.create(...)
• client.agents.sessions.create(...)