Cloud Agent - Qoder

query() 默认会在本地启动 bundled qodercli。传入 cloud agent 选项（TypeScript 的 options.experimentalCloudAgent，Python 的 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 认证（accessToken() / accessTokenFromEnv()，Python 为 access_token() / access_token_from_env()），传本机登录态或 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 { accessTokenFromEnv, query } from '@qoder-ai/qoder-agent-sdk';

const q = query({
  prompt: 'Summarize this repository in one short paragraph.',
  options: {
    auth: accessTokenFromEnv(),
    experimentalCloudAgent: {
      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: process.env.QODER_CLOUD_AGENT_ENVIRONMENT_ID!,
          title: 'first-cloud-session',
        },
      },
    },
  },
});

for await (const msg of q) {
  if (msg.type === 'result') {
    console.log('done:', msg.subtype, msg.result);
  }
}

跑完之后，从最终的 result 消息上拿到 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,
    resources: [
      { type: 'file', file_id: 'file_abc123', path: '/workspace/data.json' },
    ],
  },
}

复用已有 agent

如果 agent 已经创建好了（在控制台或前一次调用里拿到了 agent.id），传 agent: { id } 即可，不要再传 create：

experimentalCloudAgent: {
  agent: { id: 'agent_xxx' },
  session: { create: { environment_id } },
}

多轮对话：复用 session

第一轮拿到 session_id 之后，下一轮只传 session: { id }——不要再带 agent，session 自己已经绑定了 agent，混传会同步抛错。

// Turn 1: create agent + session
const first = query({
  prompt: 'My favorite color is teal. Reply with: noted.',
  options: {
    auth: accessTokenFromEnv(),
    experimentalCloudAgent: {
      agent: { create: { name: 'demo', model: 'ultimate' } },
      session: { create: { environment_id } },
    },
  },
});

let sessionId: string | undefined;
for await (const msg of first) {
  if (msg.type === 'result') sessionId = msg.session_id;
}

// Turn 2: continue the same Cloud session
const second = query({
  prompt: 'What is my favorite color?',
  options: {
    auth: accessTokenFromEnv(),
    experimentalCloudAgent: {
      session: { id: sessionId! },
    },
  },
});

for await (const msg of second) {
  if (msg.type === 'result') console.log(msg.result);  // → "teal"
}

session 上下文由 Cloud 侧保存，所以两轮 query 之间脚本可以重启、可以换机器，只要 session_id 在手就能接续。

使用 QoderSDKClient 进行多轮对话（Python）

Python 的 QoderSDKClient 提供了更高层次的 Cloud session 管理——connect() 创建/解析 Cloud session，后续 query() 逐轮复用，无需手动追踪 session_id：

import asyncio
import os

from qoder_agent_sdk import (
    QoderAgentOptions,
    QoderSDKClient,
    ResultMessage,
    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 isinstance(msg, ResultMessage):
            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 isinstance(msg, ResultMessage):
            print(msg.result)  # → "teal"
            break

    await client.disconnect()


asyncio.run(main())

注意：Cloud runtime 不支持 client.set_model()、client.reload_plugins()、MCP OAuth 等本地 CLI 控制方法，调用会抛出 ValueError。

消费 SSE 事件

Cloud runtime 通过 SSE 把 session 的事件流推回来。SDK 把每条事件包成 cloud agent 事件消息（TypeScript 为 cloud_agent_event，Python 为 CloudAgentEventMessage）：

for await (const msg of q) {
  if (msg.type === 'cloud_agent_event') {
    console.log(msg.event, msg.data);   // e.g. "user.message", "agent.message", "session.status_idle"
  } else if (msg.type === 'result') {
    // SDK synthesizes a result after receiving session.status_idle for the current turn
    console.log('turn end:', msg.subtype);
  }
}

事件结构：

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

历史事件 replay 隔离

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

SSE 调优

experimentalCloudAgent: {
  session: { id: sessionId },
  stream: {
    afterId: 'evt_xxx',         // start replay after this event ID
    deltaFlushIntervalMs: 250,  // delta merge / flush interval (SDK default if omitted)
  },
}

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

异常关闭

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

终态：result 的内容

字段	说明
`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 互斥（TypeScript 由 union 类型强制）。
已有 session.id 时不要再传 agent。
session.create 必须显式传 environment_id。
Cloud runtime 不支持本地 CLI 顶层 option：model、agent、MCP servers、settings、hooks、plugins、权限模式等传入会同步抛错。
Cloud runtime 不支持切换模型、重载插件、MCP OAuth 等运行时控制方法，只允许迭代消费消息和关闭会话（TypeScript 的 q.close()、Python 的 client.disconnect()）。

错误码

TypeScript code / Python 异常类	触发场景
`cloud_agent_auth_requires_access_token` / `CloudAgentUnsupportedAuthError`	用了本机登录态 / job token 这类非 PAT 鉴权
`cloud_agent_api_error` / `CloudAgentApiError`	Cloud OpenAPI 返回非 2xx，或 SSE 通道异常

​何时用

​前置条件

​第一次调用：创建 agent + 创建 session

​内置工具白名单

​给 session 挂载文件

​复用已有 agent

​多轮对话：复用 session

​使用 QoderSDKClient 进行多轮对话（Python）

​消费 SSE 事件

​历史事件 replay 隔离

​SSE 调优

​异常关闭

​终态：result 的内容

​用法约束速查

​错误码

​相关文档

何时用