Skip to main content
SDK

Python SDK

Install and use the Qoder Cloud Agents Python SDK, including synchronous and asynchronous calls, data types, streaming responses, retries, and HTTP configuration.

The Qoder Cloud Agents Python SDK provides synchronous and asynchronous Forward and Managed clients, with typed request parameters, Pydantic response models, automatic pagination, and SSE streaming.

Installation

Requires Python 3.10 or later: 
pip install qca-sdk

Basic usage

Save your personal access token (PAT) in an environment variable: 
export QODER_PAT="replace-with-your-PAT"
Use the Forward client to create a template and identity, start a session, and send a message: 
from qca import Forward

with Forward() as client:
    env = client.environments.create(name="demo", config={"type": "cloud"})
    identity = client.identities.create(external_id="demo-user", name="Demo User")
    assert env.id and identity.id

    template = client.templates.create(name="assistant", model="auto", environment_id=env.id)
    assert template.id
    session = client.sessions.create(identity_id=identity.id, template_id=template.id)
    assert session.id

    client.sessions.events.send(
        session.id,
        events=[{"type": "user.message", "content": [{"type": "text", "text": "Hello"}]}],
    )
    print(session.id)
Use the Managed client to create an Agent and environment, start a session, and send a message: 
from qca import Managed

with Managed() as client:
    agent = client.agents.create(name="assistant", model="auto")
    env = client.environments.create(name="demo", config={"type": "cloud"})
    assert agent.id and env.id

    session = client.sessions.create(agent=agent.id, environment_id=env.id)
    assert session.id
    client.sessions.events.send(
        session.id,
        events=[{"type": "user.message", "content": [{"type": "text", "text": "Hello"}]}],
    )
    print(session.id)

Client configuration

ConfigurationForwardManaged
Synchronous clientForwardManaged
Asynchronous clientAsyncForwardAsyncManaged
Base URL environment variableQODER_FORWARD_BASE_URLQODER_MANAGED_BASE_URL
Global default URLhttps://api.qoder.com/api/v1/forwardhttps://api.qoder.com/api/v1/cloud
CN URLhttps://api.qoder.com.cn/api/v1/forwardhttps://api.qoder.com.cn/api/v1/cloud
from qca import Forward

with Forward(
    base_url="https://api.qoder.com.cn/api/v1/forward",
    timeout=30.0,
    max_retries=2,
    default_headers={"X-Application": "my-app"},
) as client:
    result = client.models.list(timeout=5.0)
    print(len(result.data or []))
Individual requests can override timeout and extend request configuration with extra_headers, extra_query, and extra_body. with_options(...) derives a client with different settings, such as a different retry count. It shares the connection pool with the original client; closing either closes that pool.

Authentication

PAT

Set your personal access token (PAT) with the pat parameter or the QODER_PAT environment variable.

Service Account

Set QODER_SA_KEY, then explicitly create a credential component. It exchanges and caches SATs (service account tokens), refreshing them when subsequent requests require it: 
from qca import Forward, Managed, ServiceAccountCredential

with ServiceAccountCredential.from_env(
    region="global", scopes=["qca.access", "forward.access"]
) as credential:
    with Forward(
        credential=credential,
        base_url="https://api.qoder.com/api/v1/forward",
    ) as forward, Managed(
        credential=credential,
        base_url="https://api.qoder.com/api/v1/cloud",
    ) as managed:
        print(len(forward.models.list().data or []))
        print(len(managed.models.list().data or []))
Use synchronous credentials with synchronous clients, and AsyncServiceAccountCredential with asynchronous clients: 
from qca import AsyncForward, AsyncServiceAccountCredential


async def use_service_account() -> None:
    async with AsyncServiceAccountCredential.from_env(
        region="global", scopes=["qca.access", "forward.access"]
    ) as credential:
        async with AsyncForward(
            credential=credential,
            base_url="https://api.qoder.com/api/v1/forward",
        ) as client:
            result = await client.models.list()
            print(len(result.data or []))
The application owns the credential. Closing a business API client does not close the credential. Close all clients using it before closing the credential; the nested context managers above release resources in this order. Ordinary clients do not automatically read an SA Key and exchange it for a token, and an SA Key cannot serve directly as a business API access token. SA authentication failures do not fall back to PAT. If your application manages SATs itself, use Forward(auth_mode="sa", access_token=token); Managed and asynchronous clients accept the same configuration. The application is responsible for refreshing static SATs. Do not configure multiple credential sources at once. See the authentication documentation.

Asynchronous calls

Asynchronous clients use httpx.AsyncClient, with native asynchronous I/O for requests and response reads: 
import asyncio
import os

from qca import AsyncForward


async def main() -> None:
    async with AsyncForward(pat=os.environ["QODER_PAT"]) as client:
        result = await client.models.list()
        for item in result.data or []:
            print(item.id)


if __name__ == "__main__":
    asyncio.run(main())
In an already running event loop, directly await the asynchronous function. Release connections with async with or await client.close(); avoid closing a client while requests are still pending.

Request and response types

Parameters and optional fields

Request parameters use TypedDict; nested parameters can be ordinary dictionaries. Forward and Managed types are in qca.forward.types and qca.managed.types, respectively. 
from qca.forward.types import SessionUpdateParams

omitted: SessionUpdateParams = {}
explicit_null: SessionUpdateParams = {"title": None}
empty_string: SessionUpdateParams = {"title": ""}
Parameter valueRequest behavior
Omit the parameter, or pass qca.NOT_GIVEN to a parameter that supports this sentinelOmit the field
NoneSend JSON null
False, 0, ""Send as is
Empty list or dictionarySend [] or {}
Whether a field accepts null and how that value is handled are defined by the corresponding API.

Pydantic responses

Responses use Pydantic models. Read fields through attributes and serialize with to_dict() or to_json(). Unknown fields added by the server are preserved. 
result = client.sessions.retrieve("sess_example")
print(result.id)
print("title" in result.model_fields_set)
payload = result.to_dict()
model_fields_set distinguishes a missing response field from a field explicitly returned as null. The default mode constructs responses using the expected model without requiring strict validation of every field; unexpected field values may retain their original types. For strict response validation, pass _strict_response_validation=True when constructing the client. Responses that do not match the model raise APIResponseValidationError. Invalid JSON raises an error regardless of whether strict mode is enabled.

Streaming responses

Consume a single SSE connection with a context manager, which releases the connection on exit: 
with client.sessions.events.stream("sess_example") as stream:
    for event in stream:
        print(event.type)
For the asynchronous version, wait for the connection to be established before entering the asynchronous context: 
from qca import AsyncForward


async def read_events(client: AsyncForward, session_id: str) -> None:
    async with await client.sessions.events.stream(session_id) as stream:
        async for event in stream:
            print(event.type)
Ordinary stream() ends at EOF. To continue subscribing after a recoverable disconnection, use resumable_stream(): 
from qca import AsyncForward


async def read_resumable_events(
    client: AsyncForward, session_id: str, cursor: str | None = None
) -> None:
    async with client.sessions.events.resumable_stream(
        session_id, last_event_id=cursor
    ) as stream:
        async for event in stream:
            print(event.type)
            # Persist this resume position after the application successfully processes the event.
            print("cursor:", stream.last_event_id)
Asynchronous resumable_stream() directly returns an object usable with async with; do not await its creation. Synchronous clients provide the same method name, used with with and a regular for loop. A resumable stream reconnects with the SSE cursor of the most recently delivered event, without resending business requests. Without a cursor, it starts at the current tail of the event stream; reading history requires a valid historical resume position. Automatic reconnection does not guarantee exactly-once processing, and multiple deltas for one event may share an ID. HTTP retry counts and the resumable stream reconnection lifecycle are controlled separately. Set an application-level deadline for persistent subscriptions and close the stream on early exit. Stream termination alone does not establish that a business task has completed; closing a stream or cancelling a local task does not automatically stop cloud execution.

Error handling and Request ID

Request errors inherit from qca.APIError. Distinguish timeouts, connection failures, HTTP errors, and response validation failures: 
from qca import (
    APIConnectionError,
    APIResponseValidationError,
    APIStatusError,
    APITimeoutError,
)

try:
    result = client.sessions.retrieve("sess_example", timeout=10.0)
    print(result._request_id)
except APITimeoutError:
    print("Request timed out")
except APIConnectionError:
    print("Connection failed")
except APIStatusError as exc:
    print(exc.status_code, exc.code, exc.request_id)
except APIResponseValidationError:
    print("The response could not be parsed as expected")
Common HTTP exceptions include BadRequestError (400), AuthenticationError (401), PermissionDeniedError (403), NotFoundError (404), ConflictError (409), and RateLimitError (429). Use APIStatusError to handle HTTP errors together. Normal response models expose the request identifier through _request_id, and HTTP exceptions through request_id. It may be empty if the server does not return it. Asynchronous task cancellation follows the asyncio cancellation mechanism; after cleaning up resources, allow cancellation to propagate to the caller.

Retries and idempotency

By default, requests are retried up to 2 times, for a maximum of 3 attempts:
RequestRetried by default for
GET, HEAD, or requests with an idempotency keyConnection errors, 408, 429, 5xx
Other requests without an idempotency key429
409 responseNo automatic retries
Requests must also meet constraints such as safe replayability; some APIs apply stricter rules. Within these rules, the SDK handles server retry hints and uses valid server-specified delays or exponential backoff. Set max_retries=0 when constructing a client to disable retries. Use client.with_options(max_retries=1) to adjust retries for a derived client. HTTP retry configuration does not automatically resume an established ordinary SSE connection; explicitly select a resumable stream. Write methods that support idempotency keys provide an idempotency_key parameter. You can also pass the key through extra_headers={"Idempotency-Key": key}. Retries of the same logical operation must reuse the same key.

Timeouts and cancellation

The default connection timeout is 5 seconds. Read, write, and connection pool wait timeouts are each 600 seconds. These are per-phase timeouts for each attempt, not a total deadline including retries and waiting. Pass a number of seconds, an httpx.Timeout, or None to disable timeouts: 
import httpx

result = client.sessions.retrieve(
    "sess_example",
    timeout=httpx.Timeout(30.0, connect=5.0),
)
To limit the total duration of an asynchronous call or subscription, use asyncio.wait_for. The following function reuses read_resumable_events above: 
import asyncio

from qca import AsyncForward


async def read_for_one_minute(client: AsyncForward, session_id: str) -> None:
    try:
        await asyncio.wait_for(
            read_resumable_events(client, session_id), timeout=60.0
        )
    except asyncio.TimeoutError:
        print("The subscription reached the application-defined deadline")
The application must manage the overall deadline for synchronous calls. For SSE, the read timeout limits the wait for the next chunk of data; it does not mean that the stream ends after 600 seconds in total.

Pagination

List responses can be iterated across pages directly; .data contains only the current page: 
page = client.sessions.list(limit=20)
print(len(page.data))

for item in page:
    print(item.id)
Use iter_pages() to process pages, or check has_next_page() before calling get_next_page(). Asynchronous clients support async for directly for pagination: 
from qca import AsyncForward


async def list_all(client: AsyncForward) -> None:
    async for item in client.sessions.list(limit=20):
        print(item.id)
Paginators preserve filters and request configuration. Not every collection endpoint is paginated; read .data directly for unpaginated responses.

Advanced configuration

Raw responses and response body reads

with_raw_response returns the status code, response headers, and a parsable response body: 
raw = client.models.with_raw_response.list()
print(raw.status_code, raw.headers.get("x-request-id"))
result = raw.parse()
To read headers before deciding how to consume the response body, use with_streaming_response and release the connection through a context manager: 
with client.models.with_streaming_response.list() as response:
    print(response.headers.get("x-request-id"))
    result = response.parse()
This controls how the HTTP response body is read and is a different interface from SSE event subscriptions. The asynchronous version uses async with and await response.parse().

Custom HTTP clients

Pass an httpx.Client or httpx.AsyncClient to configure proxies, TLS, transport, and connection pools: 
import httpx

from qca import Forward

with Forward(
    http_client=httpx.Client(
        limits=httpx.Limits(max_connections=50, max_keepalive_connections=20)
    )
) as client:
    print(len(client.models.list().data or []))
Use extra_headers, extra_query, and extra_body to extend headers, query parameters, and the body. These settings can override fields set by the SDK, including authentication headers, so override only fields your application explicitly needs to adjust.