Python SDK

Qoder Cloud Agents Python SDK は、同期・非同期の Forward、Managed クライアントを提供し、型付きリクエストパラメーター、Pydantic レスポンスモデル、自動ページネーション、SSE ストリーミング読み取りを備えています。

インストール

Python 3.10 以降が必要です：

pip install qca-sdk

基本的な使い方

個人アクセストークン（PAT）を環境変数に保存します：

export QODER_PAT="替换为你的 PAT"

Forward クライアントでテンプレートと ID を作成し、セッションを開始してメッセージを送信します：

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 のデフォルト URL	`https://api.qoder.com/api/v1/forward`	`https://api.qoder.com/api/v1/cloud`
CN の URL	`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 []))

認証情報はアプリケーションが所有します。業務 API クライアントを閉じても認証情報は閉じられません。それを使用する全クライアントを閉じてから認証情報を閉じてください。上記のネストしたコンテキストマネージャーはこの順序でリソースを解放します。通常のクライアントは SA Key を自動で読み取ってトークンに交換しません。SA Key を業務 API のアクセストークンとして直接使用することもできません。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 を受け付けるか、その値をどう処理するかは、対応する API の定義に従います。

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 カーソルで再接続し、業務リクエストを再送しません。カーソルがなければ現在のイベントストリーム末尾から開始します。履歴の読み取りには有効な過去の再開位置が必要です。自動再接続は「厳密に 1 回」の処理を保証せず、同じイベントの複数の差分が同一 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 レスポンス	自動リトライしない

リクエストは安全な再実行などの制約も満たす必要があります。一部 API にはより厳しいルールがあります。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)

ページネーターはフィルター条件とリクエスト設定を保持します。すべてのコレクション API がページネーション対応とは限りません。ページ分割のないレスポンスは .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 クライアント