> ## Documentation Index
> Fetch the complete documentation index at: https://docs.qoder.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Cloud Agent（experimental）

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

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

<div id="何时用" />

## 何时用

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

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

<div id="前置条件" />

## 前置条件

* **Personal Access Token (PAT)**：到 [qoder.com/account/integrations](https://qoder.com/account/integrations) 生成，详见 [SDK 认证](/zh/cli/sdk/python/authentication)。Cloud runtime 只接受 `access_token()` / `access_token_from_env()`，传 `qodercli_auth()` / `job_token()` 会同步报错。
* **Cloud environment\_id**：新建 session 必须指定容器环境 ID。从 Qoder 控制台或管理 API 获取。

```bash theme={null}
export QODER_PERSONAL_ACCESS_TOKEN="<your-pat>"
export QODER_CLOUD_AGENT_ENVIRONMENT_ID="<your-env-id>"
```

<div id="第一次调用创建-agent-创建-session" />

<div id="第一次调用创建agent创建session" />

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

最常见的入门路径——一次性创建一个新的 Cloud Agent，并立刻为它开一个 session 来跑 prompt：

```python theme={null}
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)）。

<div id="内置工具白名单" />

### 内置工具白名单

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

<div id="给-session-挂载文件" />

<div id="给session挂载文件" />

### 给 session 挂载文件

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

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

<div id="复用已有-agent" />

<div id="复用已有agent" />

## 复用已有 agent

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

```python theme={null}
experimental_cloud_agent={
    "agent": {"id": "agent_xxx"},
    "session": {"create": {"environment_id": environment_id}},
}
```

<div id="多轮对话复用-session" />

<div id="多轮对话复用session" />

## 多轮对话：复用 session

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

```python theme={null}
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` 在手就能接续。

<div id="使用-qodersdkclient-进行多轮对话" />

<div id="使用qodersdkclient进行多轮对话" />

## 使用 QoderSDKClient 进行多轮对话

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

```python theme={null}
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

    # 第二轮：直接调用 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 runtime 不支持 `client.set_model()`、`client.reload_plugins()`、MCP OAuth 等本地 CLI 控制方法，调用会抛出 `ValueError`。

<div id="消费-sse-事件" />

<div id="消费sse事件" />

## 消费 SSE 事件

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

```python theme={null}
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 在收到当前 turn 的 session.status_idle 后合成 ResultMessage
        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                                               |

<div id="历史事件-replay-隔离" />

<div id="历史事件replay隔离" />

### 历史事件 replay 隔离

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

<div id="sse-调优" />

<div id="sse调优" />

### SSE 调优

```python theme={null}
experimental_cloud_agent={
    "session": {"id": session_id},
    "stream": {
        "after_id": "evt_xxx",            # 从指定事件 ID 之后开始 replay
        "delta_flush_interval_ms": 250,   # 增量内容的合并/刷新间隔（默认由 SDK 决定）
    },
}
```

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

<div id="异常关闭" />

### 异常关闭

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

<div id="终态resultmessage-的内容" />

<div id="终态resultmessage的内容" />

## 终态：`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` 回填 |

<div id="用法约束速查" />

## 用法约束速查

* `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()`。

<div id="错误码" />

## 错误码

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

<div id="相关文档" />

## 相关文档

* [SDK 认证](/zh/cli/sdk/python/authentication) — PAT 获取与环境变量
* [多轮对话](/zh/cli/sdk/python/multi-turn-conversation) — 本地 runtime 的多轮对话方式
* [SDK References](/zh/cli/sdk/python/references) — `QoderAgentOptions.experimental_cloud_agent`、`CloudAgentEventMessage` 完整字段