Cloud Agent（experimental）

query() 默认会在本地启动 bundled qodercli。传入 options.experimental_cloud_agent 后，SDK 改走 Qoder Cloud Agent runtime——agent 和 session 跑在 Qoder Cloud 容器里，本地只负责发请求和消费 SSE 事件流。

状态：experimental / unstable。API 形状可能在小版本之间变化；不要在生产链路上依赖未标稳定的字段。

​

何时用

• 不想在本机管理 qodercli、bundled 二进制、运行环境
• 需要长驻 / 跨机器复用同一个 agent 实例（agent 由 Cloud 持久化）
• 需要把会话上下文留在 Cloud 上，多端 / 多次脚本调用复用同一 session

本地 CLI 才有的能力——mcp_servers / settings / hooks / plugins / 本地 permissions / checkpoint 等——在 Cloud runtime 下都不支持，传入会同步抛错。

​

前置条件

• Personal Access Token (PAT)：到 qoder.com/account/integrations 生成，详见 SDK 认证。Cloud runtime 只接受 access_token() / access_token_from_env()，传 qodercli_auth() / job_token() 会同步报错。
• Cloud environment_id：新建 session 必须指定容器环境 ID。从 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（见 多轮对话）。

​

内置工具白名单

tools[].enabled_tools 当前支持：bash、write、glob、web_fetch、read、edit、grep、web_search。不传 tools 表示该 agent 没有工具。

​

给 session 挂载文件

通过 Files API 上传文件后，可以在 session.create.resources 里挂载到容器：

"session": {
    "create": {
        "environment_id": environment_id,
        "resources": [
            {"type": "file", "file_id": "file_abc123", "path": "/workspace/data.json"},
        ],
    },
}

​

复用已有 agent

如果 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: create 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: continue the same 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 侧保存，所以两轮 query 之间脚本可以重启、可以换机器，只要 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() creates the Cloud session; optionally pass a first-turn prompt
    await client.connect("My favorite color is teal. Reply with: noted.")

    # Consume first turn messages
    async for msg in client.receive_messages():
        if msg.get("type") == "result":
            break

    # Turn 2: call query() directly — session is already bound
    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 runtime 不支持 client.set_model()、client.reload_plugins()、MCP OAuth 等本地 CLI 控制方法，调用会抛出 ValueError。

​

消费 SSE 事件

Cloud runtime 通过 SSE 把 session 的事件流推回来。SDK 把每条事件包成 CloudAgentEventMessage 消息：

from qoder_agent_sdk import CloudAgentEventMessage

async for msg in result:
    if isinstance(msg, CloudAgentEventMessage):
        print(msg.event, msg.data)  # e.g. "user.message", "agent.message", "session.status_idle"
    elif hasattr(msg, "subtype"):
        # SDK synthesizes a ResultMessage after receiving session.status_idle for the current turn
        print("turn end:", msg.subtype)

事件结构（CloudAgentEventMessage 字段）：

字段	说明
event	Cloud 事件名（如 user.message、agent.message、session.status_idle）
id	事件在 SSE 流里的 ID，可用作 replay 起点
data	Cloud 事件 payload（含 turn_id 等字段）
uuid	SDK 内部生成的唯一 ID，用于去重
session_id	所属 Cloud session ID

​

历史事件 replay 隔离

复用已有 session 时，SSE 会先 replay 历史 turn 的事件——SDK 内部按 turn_id 隔离：只有当前 turn 的 session.status_idle 才会触发 ResultMessage 终态，历史事件不会让你的 query 提前结束。

​

SSE 调优

experimental_cloud_agent={
    "session": {"id": session_id},
    "stream": {
        "after_id": "evt_xxx",            # start replay after this event ID
        "delta_flush_interval_ms": 250,   # delta merge / flush interval (SDK default if omitted)
    },
}

兼容性：afterId / deltaFlushIntervalMs（camelCase）也被接受，运行时会自动识别。

​

异常关闭

如果当前 turn 还没收到终态，SSE 就先断开了，SDK 会合成一条 error ResultMessage（subtype != 'success'，is_error=True），方便上层统一处理。

​

终态：ResultMessage 的内容

字段	说明
subtype	"success" 或错误子类型
is_error	布尔，是否异常结束
session_id	Cloud session ID（创建分支时由 SDK 回填）
result	agent 当轮的文本回复（多 text block 会拼接）
usage / model_usage / total_cost_usd	从当前 turn 的 span.model_request_end.usage 回填

​

用法约束速查

• agent 和 session 各自的 id / create 互斥。
• 已有 session["id"] 时不要再传 agent。
• session["create"] 必须显式传 environment_id。
• Cloud runtime 不支持本地 CLI 顶层 option：model、agent、mcp_servers、settings、hooks、plugins、permission_mode 等传入会同步抛错。
• Cloud runtime 不支持 QoderSDKClient.set_model()、reload_plugins()、MCP OAuth 等控制方法，只允许 async for 消费消息和 close()。

​

错误码

异常类	触发场景
CloudAgentUnsupportedAuthError	用了 qodercli_auth() / job_token() 这类非 PAT 鉴权
CloudAgentApiError	Cloud OpenAPI 返回非 2xx，或 SSE 通道异常

​

相关文档

• SDK 认证 — PAT 获取与环境变量
• 多轮对话 — 本地 runtime 的多轮对话方式
• API References — QoderAgentOptions.experimental_cloud_agent、CloudAgentEventMessage 完整字段