メインコンテンツへスキップ
query() はデフォルトでバンドル済みの qodercli をローカルで起動します。options.experimental_cloud_agent を渡すと SDK は Qoder Cloud Agent ランタイムに切り替わります — agent と session は Qoder Cloud のコンテナ内で実行され、ローカルプロセスはリクエスト送信と SSE イベントストリームの消費だけを担当します。
ステータス：experimental / unstable。API 形状はマイナーバージョン間で変化する可能性があります。安定化されていないフィールドを本番のコードパスで依存しないでください。

​
使うべきタイミング

  • qodercli、バンドル済みバイナリ、ローカルランタイムを管理したくない
  • 同じ agent インスタンスをマシン間で長期間再利用したい（agent は Cloud に永続化される）
  • session のコンテキストを Cloud に置いて、複数のプロセス / ホストから再開したい
ローカル CLI 専用の機能 — mcp_servers / settings / hooks / plugins / ローカル permissions / checkpoint — は Cloud ランタイムではサポートされません。これらを渡すと同期的にエラーが投げられます。

​
前提条件

  • Personal Access Token (PAT)：qoder.com/account/integrations で生成。詳細は SDK 認証 を参照。Cloud ランタイムは access_token() / access_token_from_env() のみを受け付けます。qodercli_auth() / job_token() を渡すと同期的にエラーになります。
  • Cloud environment_id：session 作成時に必須。Qoder コンソールまたは管理 API から取得します。
export QODER_PERSONAL_ACCESS_TOKEN="<your-pat>"
export QODER_CLOUD_AGENT_ENVIRONMENT_ID="<your-env-id>"

​
初回呼び出し：agent 作成 + session 作成

最も一般的なエントリパス — 新しい Cloud Agent を作成し、すぐにその session を開いて prompt を実行します： 
import asyncio
import os

from qoder_agent_sdk import QoderAgentOptions, access_token_from_env, query


async def main():
    result = await query(
        prompt="Summarize this repository in one short paragraph.",
        options=QoderAgentOptions(
            auth=access_token_from_env(),
            experimental_cloud_agent={
                "agent": {
                    "create": {
                        "name": "my-cloud-agent",
                        "model": "ultimate",
                        "system": "You are a concise code assistant.",
                        "tools": [
                            {
                                "type": "agent_toolset_20260401",
                                "enabled_tools": ["read", "glob", "grep"],
                            },
                        ],
                    },
                },
                "session": {
                    "create": {
                        "environment_id": os.environ["QODER_CLOUD_AGENT_ENVIRONMENT_ID"],
                        "title": "first-cloud-session",
                    },
                },
            },
        ),
    )

    async for msg in result:
        if hasattr(msg, "subtype") and msg.subtype == "success":
            print("done:", msg.result)


asyncio.run(main())
ターン終了時、最後の ResultMessage から session_id を取得します — 後続のターンでこれを使って同じ session を再開します（マルチターン：session の再開 を参照）。

​
ビルトインツールの許可リスト

tools[].enabled_tools で現在サポートされているもの：bash、write、glob、web_fetch、read、edit、grep、web_search。tools を省略すると agent はツールを持ちません。

​
session へのファイルマウント

Files API でアップロードしたファイルを session.create.resources で session コンテナにマウントできます： 
"session": {
    "create": {
        "environment_id": environment_id,
        "resources": [
            {"type": "file", "file_id": "file_abc123", "path": "/workspace/data.json"},
        ],
    },
}

​
既存 agent の再利用

すでに agent.id がある場合（コンソール経由または前回の呼び出しで取得）、agent: {"id": ...} を渡し create は省略します： 
experimental_cloud_agent={
    "agent": {"id": "agent_xxx"},
    "session": {"create": {"environment_id": environment_id}},
}

​
マルチターン：session の再開

初回ターンから session_id を得たら、次の呼び出しは**session: {"id": ...} のみ**を渡します — agent は含めないでください。session は既に agent を束縛しており、両方を渡すと同期的にエラーになります。 
import asyncio
import os

from qoder_agent_sdk import QoderAgentOptions, access_token_from_env, query


async def main():
    environment_id = os.environ["QODER_CLOUD_AGENT_ENVIRONMENT_ID"]

    # Turn 1: agent + session を作成
    session_id = None
    first = await query(
        prompt="My favorite color is teal. Reply with: noted.",
        options=QoderAgentOptions(
            auth=access_token_from_env(),
            experimental_cloud_agent={
                "agent": {"create": {"name": "demo", "model": "ultimate"}},
                "session": {"create": {"environment_id": environment_id}},
            },
        ),
    )

    async for msg in first:
        if hasattr(msg, "session_id") and hasattr(msg, "subtype"):
            session_id = msg.session_id

    # Turn 2: 同じ Cloud session を続行
    second = await query(
        prompt="What is my favorite color?",
        options=QoderAgentOptions(
            auth=access_token_from_env(),
            experimental_cloud_agent={
                "session": {"id": session_id},
            },
        ),
    )

    async for msg in second:
        if hasattr(msg, "result") and msg.result:
            print(msg.result)  # → "teal"


asyncio.run(main())
session のコンテキストは Cloud 側に保存されるため、ターン間でスクリプトを再起動したりマシンを変えたりしても、session_id さえあれば再開できます。

​
QoderSDKClient でのマルチターン会話

QoderSDKClient はより高レベルの Cloud session 管理を提供します — connect() で Cloud session を作成/解決し、後続の query() はターンごとに再利用します。session_id を手動で追跡する必要はありません： 
import asyncio
import os

from qoder_agent_sdk import QoderAgentOptions, QoderSDKClient, access_token_from_env


async def main():
    environment_id = os.environ["QODER_CLOUD_AGENT_ENVIRONMENT_ID"]

    client = QoderSDKClient(
        options=QoderAgentOptions(
            auth=access_token_from_env(),
            experimental_cloud_agent={
                "agent": {"create": {"name": "demo", "model": "ultimate"}},
                "session": {"create": {"environment_id": environment_id}},
            },
        )
    )

    # connect() で Cloud session を作成。オプションで初回ターンの prompt を渡す
    await client.connect("My favorite color is teal. Reply with: noted.")

    # 初回ターンのメッセージを消費
    async for msg in client.receive_messages():
        if msg.get("type") == "result":
            break

    # Turn 2: query() を直接呼ぶ — session は既に束縛済み
    await client.query("What is my favorite color?")
    async for msg in client.receive_messages():
        if msg.get("type") == "result":
            print(msg.get("result"))  # → "teal"
            break

    await client.close()


asyncio.run(main())
注意：Cloud ランタイムは client.set_model()、client.reload_plugins()、MCP OAuth などのローカル CLI 制御メソッドをサポートしません。呼び出すと ValueError が投げられます。

​
SSE イベントの消費

Cloud ランタイムは SSE 経由で session のイベントストリームをプッシュバックします。SDK は各イベントを CloudAgentEventMessage メッセージとしてラップします： 
from qoder_agent_sdk import CloudAgentEventMessage

async for msg in result:
    if isinstance(msg, CloudAgentEventMessage):
        print(msg.event, msg.data)  # 例：「user.message」「agent.message」「session.status_idle」
    elif hasattr(msg, "subtype"):
        # SDK は現在ターンの session.status_idle 受信後に ResultMessage を合成
        print("turn end:", msg.subtype)
イベント構造（CloudAgentEventMessage フィールド）：
フィールド説明
eventCloud イベント名（例：user.message、agent.message、session.status_idle）
idSSE ストリーム内のイベント ID。replay の起点として使える
dataCloud イベント payload（turn_id などのフィールドを含む）
uuidSDK が内部生成する一意 ID。重複排除に使用
session_idこのイベントが属する Cloud session ID

​
履歴イベントの replay 隔離

既存 session を再開する際、SSE はまず履歴ターンのイベントを replay します。SDK は turn_id で隔離します — 現在のターンの session.status_idle のみが ResultMessage 終端をトリガーし、履歴イベントが query を早期終了させることはありません。

​
SSE チューニング

experimental_cloud_agent={
    "session": {"id": session_id},
    "stream": {
        "after_id": "evt_xxx",            # 指定イベント ID 以降から replay を開始
        "delta_flush_interval_ms": 250,   # デルタ内容のマージ / フラッシュ間隔（省略時は SDK デフォルト）
    },
}
互換性：afterId / deltaFlushIntervalMs（camelCase）もランタイムで受け付けられます。

​
異常終了

現在ターンが終端イベントに達する前に SSE が切断された場合、SDK はエラー ResultMessage を合成します（subtype != 'success'、is_error=True）。呼び出し側は統一的に処理できます。

​
ResultMessage 終端

フィールド説明
subtype"success" または各種エラー subtype
is_errorboolean。異常終了かどうか
session_idCloud session ID（create 分岐時は SDK が回填）
resultこのターンの agent テキスト返答（複数 text block は連結される）
usage / model_usage / total_cost_usd現在ターンの span.model_request_end.usage から回填

​
制約サマリ

  • agent と session の id / create はそれぞれ相互排他。
  • 既存 session["id"] を渡すとき、agent は渡してはなりません。
  • session["create"] には environment_id を明示的に渡す必要があります。
  • Cloud ランタイムはローカル CLI 専用のトップレベル option を拒否します：model、agent、mcp_servers、settings、hooks、plugins、permission_mode など。渡すと同期的にエラーが投げられます。
  • Cloud ランタイムは QoderSDKClient.set_model()、reload_plugins()、MCP OAuth などをサポートしません。async for 消費と close() のみが保証されます。

​
エラーコード

例外クラス発生条件
CloudAgentUnsupportedAuthError非 PAT 認証（qodercli_auth() / job_token() など）が使用された
CloudAgentApiErrorCloud OpenAPI が non-2xx を返したか、SSE チャネルが異常終了した

​
関連ドキュメント

  • SDK 認証 — PAT の取得と環境変数
  • マルチターン会話 — ローカルランタイム下のマルチターン
  • SDK References — QoderAgentOptions.experimental_cloud_agent と CloudAgentEventMessage の完全フィールド