Python SDK

Qoder Cloud Agents Python SDK 提供同步和异步的 Forward、Managed 客户端，包含类型化请求参数、Pydantic 响应模型、自动分页和 SSE 流式读取。

安装

需要 Python 3.10 或更高版本：

pip install qca-sdk

基本用法

将个人访问令牌（PAT）保存到环境变量：

export QODER_PAT="替换为你的 PAT"

使用 Forward 客户端创建模板和身份，发起会话并发送消息：

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": "你好"}]}],
    )
    print(session.id)

使用 Managed 客户端创建 Agent 和环境，发起会话并发送消息：

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": "你好"}]}],
    )
    print(session.id)

客户端配置

配置	Forward	Managed
同步客户端	`Forward`	`Managed`
异步客户端	`AsyncForward`	`AsyncManaged`
Base URL 环境变量	`QODER_FORWARD_BASE_URL`	`QODER_MANAGED_BASE_URL`
Global 默认地址	`https://api.qoder.com/api/v1/forward`	`https://api.qoder.com/api/v1/cloud`
CN 地址	`https://api.qoder.com.cn/api/v1/forward`	`https://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 []))

单次请求可覆盖 timeout，并通过 extra_headers、extra_query、extra_body 补充请求配置。with_options(...) 可派生不同配置的客户端，例如调整重试次数；它与原客户端共享连接池，关闭任意一方都会关闭该连接池。

鉴权

PAT

通过 pat 参数或 QODER_PAT 环境变量设置个人访问令牌（PAT）。

Service Account

设置 QODER_SA_KEY，再显式创建凭证组件。组件置换并缓存 SAT（服务账号令牌），在后续请求需要时更新：

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 []))

同步客户端配合同步凭证，异步客户端使用 AsyncServiceAccountCredential：

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 []))

凭证由应用持有，关闭业务客户端不会关闭凭证。先关闭所有使用它的客户端，再关闭凭证；上面的嵌套上下文管理器按这个顺序释放资源。普通客户端不会自动读取 SA Key 并换取令牌，SA Key 也不能直接作为业务接口的访问令牌。SA 鉴权失败不会回退到 PAT。如果应用自行管理 SAT，使用 Forward(auth_mode="sa", access_token=token)；Managed 和异步客户端接受同样的配置。静态 SAT 的更新由应用负责，不要同时配置多个凭证来源。更多说明见认证文档。

异步调用

异步客户端基于 httpx.AsyncClient，请求和响应读取使用原生异步 I/O：

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())

在已经运行的事件循环中直接 await 异步函数。异步客户端使用 async with 或 await client.close() 释放连接；避免在请求尚未完成时关闭客户端。

请求与响应类型

参数和可选字段

请求参数使用 TypedDict，嵌套参数可直接传入普通字典。Forward 和 Managed 的类型分别位于 qca.forward.types 和 qca.managed.types。

from qca.forward.types import SessionUpdateParams

omitted: SessionUpdateParams = {}
explicit_null: SessionUpdateParams = {"title": None}
empty_string: SessionUpdateParams = {"title": ""}

参数写法	请求行为
省略参数，或为支持该哨兵值的参数传入 `qca.NOT_GIVEN`	不发送该字段
`None`	发送 JSON `null`
`False`、`0`、`""`	原样发送
空列表、空字典	发送 `[]`、`{}`

字段是否接受 null，以及如何处理该值，以对应接口定义为准。

Pydantic 响应

响应使用 Pydantic 模型，可以通过属性读取字段，通过 to_dict()、to_json() 序列化。服务端新增的未知字段会保留。

result = client.sessions.retrieve("sess_example")
print(result.id)
print("title" in result.model_fields_set)
payload = result.to_dict()

model_fields_set 可区分“响应中没有该字段”和“响应显式返回 null”。默认模式按预期模型构造响应，但不要求每个字段都通过严格校验；异常字段值可能保留原类型。需要严格响应校验时，构造客户端时传入 _strict_response_validation=True。不符合模型定义的响应会抛出 APIResponseValidationError；无论是否启用严格模式，无效 JSON 都会报错。

流式响应

通过上下文管理器消费单次 SSE 连接，退出时释放连接：

with client.sessions.events.stream("sess_example") as stream:
    for event in stream:
        print(event.type)

异步版本需要先等待连接建立，再进入异步上下文：

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)

普通 stream() 在 EOF 时结束。需要在可恢复的断连后继续订阅时，使用 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)
            # 应用成功处理事件后，再持久化该恢复位置。
            print("cursor:", stream.last_event_id)

异步 resumable_stream() 直接返回可用于 async with 的对象，创建它时不加 await。同步客户端提供同名方法，使用 with 和普通 for。恢复流携带最近交付事件的 SSE 游标重新连接，不重新发送业务请求。没有游标时从当前事件流尾部开始；补读历史需要有效的历史恢复位置。自动重连不保证“恰好一次”处理，同一事件的多个增量也可能共用 ID。 HTTP 重试次数与恢复流的重连生命周期分开控制。为持续订阅设置应用层期限，并在提前退出时关闭流。流结束不能单独用来判断业务任务已经完成；关闭流或取消本地任务也不会自动停止云端执行。

错误处理与 Request ID

请求错误继承自 qca.APIError。区分超时、连接失败、HTTP 错误与响应校验失败：

from qca import (
    APIConnectionError,
    APIResponseValidationError,
    APIStatusError,
    APITimeoutError,
)

try:
    result = client.sessions.retrieve("sess_example", timeout=10.0)
    print(result._request_id)
except APITimeoutError:
    print("请求超时")
except APIConnectionError:
    print("连接失败")
except APIStatusError as exc:
    print(exc.status_code, exc.code, exc.request_id)
except APIResponseValidationError:
    print("响应无法按预期解析")

常见 HTTP 异常包括 BadRequestError（400）、AuthenticationError（401）、PermissionDeniedError（403）、NotFoundError（404）、ConflictError（409）、RateLimitError（429）。使用 APIStatusError 可以统一处理 HTTP 错误。正常响应模型通过 _request_id 提供请求标识，HTTP 异常通过 request_id 提供。服务端未返回时可能为空。异步任务取消遵循 asyncio 的取消机制，清理资源后应允许取消继续向调用方传播。

重试与幂等性

默认最多重试 2 次，即最多尝试 3 次：

请求	默认可重试情况
GET、HEAD，或携带幂等键的请求	连接错误、408、429、5xx
未携带幂等键的其他请求	429
409 响应	不自动重试

请求还需满足可安全重放等限制；部分接口有更严格的规则。SDK 在这些规则内处理服务端重试提示，使用服务端指定的有效等待时间或指数退避。构造客户端时设置 max_retries=0 可关闭重试，client.with_options(max_retries=1) 可调整派生客户端的重试次数。已建立的普通 SSE 连接不会因为 HTTP 重试配置自动恢复，应显式选择恢复流。支持幂等键的写方法提供 idempotency_key 参数，也可通过 extra_headers={"Idempotency-Key": key} 传入。同一次逻辑操作的重试必须复用同一个键。

超时与取消

默认连接超时为 5 秒，读取、写入和连接池等待超时各为 600 秒。这些是每次尝试的分阶段超时，不是包含重试和等待时间的总期限。可以传入秒数、httpx.Timeout，或使用 None 关闭超时：

import httpx

result = client.sessions.retrieve(
    "sess_example",
    timeout=httpx.Timeout(30.0, connect=5.0),
)

需要控制异步调用或订阅的总时长时，使用 asyncio.wait_for。以下函数复用前面的 read_resumable_events：

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("订阅已达到应用设置的期限")

同步调用的整体期限需要应用自行管理。对 SSE 而言，读取超时限制等待下一段数据的时间，并不意味着流会在总计 600 秒后结束。

分页

列表响应可直接跨页迭代；.data 只包含当前页：

page = client.sessions.list(limit=20)
print(len(page.data))

for item in page:
    print(item.id)

使用 iter_pages() 按页处理，或先检查 has_next_page() 再调用 get_next_page()。异步客户端的分页可以直接使用 async for：

from qca import AsyncForward


async def list_all(client: AsyncForward) -> None:
    async for item in client.sessions.list(limit=20):
        print(item.id)

分页器保留筛选条件及请求配置。不是所有集合接口都分页；无分页的响应直接读取 .data。

高级配置

原始响应和响应体读取

with_raw_response 返回状态码、响应头和可解析的响应体：

raw = client.models.with_raw_response.list()
print(raw.status_code, raw.headers.get("x-request-id"))
result = raw.parse()

需要先读取响应头、再决定如何消费响应体时，使用 with_streaming_response，并通过上下文管理器释放连接：

with client.models.with_streaming_response.list() as response:
    print(response.headers.get("x-request-id"))
    result = response.parse()

它控制 HTTP 响应体的读取方式，与 SSE 事件订阅是不同的接口。异步版本使用 async with 和 await response.parse()。

自定义 HTTP 客户端

传入 httpx.Client 或 httpx.AsyncClient，配置代理、TLS、Transport 和连接池：

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 []))

扩展 Header、Query 和 Body 时，使用 extra_headers、extra_query、extra_body。这些配置可覆盖 SDK 已设置的字段，包括认证头，因此只覆盖应用明确需要调整的字段。

​安装

​基本用法

​客户端配置

​鉴权

​PAT

​Service Account

​异步调用

​请求与响应类型

​参数和可选字段

​Pydantic 响应

​流式响应

​错误处理与 Request ID

​重试与幂等性

​超时与取消

​分页

​高级配置

​原始响应和响应体读取

​自定义 HTTP 客户端

安装