> ## 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) By default, `query()` launches the bundled qodercli locally. Pass `options.experimental_cloud_agent` and the SDK switches to the Qoder Cloud Agent runtime instead — the agent and session run in a Qoder Cloud container, while the local process only sends requests and consumes the SSE event stream. > **Status**: experimental / unstable. The API shape may change between minor versions; do not depend on unreleased fields in production code paths.
## When to use * You don't want to manage qodercli, the bundled binary, or a local runtime * You need a long-lived agent reused across machines (the agent is persisted in the Cloud) * You want session context to live in the Cloud so multiple processes / hosts can resume it Local-CLI-only capabilities — `mcp_servers` / `settings` / `hooks` / `plugins` / local permissions / checkpoint — are **not** available under the Cloud runtime; passing any of them throws synchronously.
## Prerequisites * **Personal Access Token (PAT)**: generated at [qoder.com/account/integrations](https://qoder.com/account/integrations); see [SDK Authentication](/en/cli/sdk/python/authentication). The Cloud runtime accepts only `access_token()` / `access_token_from_env()`. Passing `qodercli_auth()` / `job_token()` throws synchronously. * **Cloud `environment_id`**: required when creating a session. Get it from the Qoder console or the management API. ```bash theme={null} export QODER_PERSONAL_ACCESS_TOKEN="" export QODER_CLOUD_AGENT_ENVIRONMENT_ID="" ```
## First call: create agent + create session The most common entry path — create a new Cloud Agent and immediately open a session for it to run a 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()) ``` When the turn finishes, read `session_id` from the final `ResultMessage` — later turns use it to resume the same session (see [Multi-turn: resuming a session](#multi-turn-resuming-a-session)).
### Built-in tool allowlist `tools[].enabled_tools` currently supports: `bash`, `write`, `glob`, `web_fetch`, `read`, `edit`, `grep`, `web_search`. Omit `tools` to give the agent no tools.
### Mounting files into a session After uploading a file via the Files API, mount it into the session container with `session.create.resources`: ```python theme={null} "session": { "create": { "environment_id": environment_id, "resources": [ {"type": "file", "file_id": "file_abc123", "path": "/workspace/data.json"}, ], }, } ```
## Reusing an existing agent If you already have an `agent.id` (created via the console or a previous call), pass `agent: {"id": ...}` and skip `create`: ```python theme={null} experimental_cloud_agent={ "agent": {"id": "agent_xxx"}, "session": {"create": {"environment_id": environment_id}}, } ```
## Multi-turn: resuming a session Once you have a `session_id` from the first turn, the next call passes **only `session: {"id": ...}`** — do **not** include `agent`. The session already binds an agent, and combining the two throws synchronously. ```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 context lives in the Cloud, so the script can restart or move between machines between turns — as long as you have the `session_id`, you can resume.
## Using QoderSDKClient for multi-turn conversations `QoderSDKClient` provides a higher-level Cloud session management — `connect()` creates/resolves the Cloud session, subsequent `query()` calls reuse it per turn, without needing to manually track `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() 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()) ``` > **Note**: The Cloud runtime does not support `client.set_model()`, `client.reload_plugins()`, MCP OAuth, or other local-CLI control methods — calling them raises `ValueError`.
## Consuming SSE events The Cloud runtime streams session events back over SSE. The SDK wraps each event as a `CloudAgentEventMessage` message: ```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 synthesizes a ResultMessage after receiving session.status_idle for the current turn print("turn end:", msg.subtype) ``` Event shape (`CloudAgentEventMessage` fields): | Field | Description | | ------------ | ------------------------------------------------------------------------------ | | `event` | Cloud event name (e.g. `user.message`, `agent.message`, `session.status_idle`) | | `id` | Event ID in the SSE stream; usable as a replay anchor | | `data` | Cloud event payload (includes `turn_id` and other fields) | | `uuid` | SDK-generated unique ID for deduplication | | `session_id` | Cloud session ID this event belongs to |
### History replay isolation When resuming an existing session, the SSE stream first replays history events. The SDK isolates by `turn_id`: only the **current turn**'s `session.status_idle` triggers the `ResultMessage` terminal — historical events will not end your query early.
### SSE tuning ```python theme={null} 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) }, } ``` > **Compatibility**: `afterId` / `deltaFlushIntervalMs` (camelCase) are also accepted at runtime.
### Abnormal close If SSE disconnects before the current turn reaches a terminal event, the SDK synthesizes an error `ResultMessage` (`subtype != 'success'`, `is_error=True`) so callers can handle it uniformly.
## The `ResultMessage` terminal | Field | Description | | ------------------------------------------ | ----------------------------------------------------------------------- | | `subtype` | `"success"` or an error subtype | | `is_error` | Boolean; whether the turn ended abnormally | | `session_id` | Cloud session ID (backfilled by the SDK on the create branch) | | `result` | Agent's text reply for the turn (multiple text blocks are concatenated) | | `usage` / `model_usage` / `total_cost_usd` | Backfilled from the current turn's `span.model_request_end.usage` |
## Constraints at a glance * `agent` and `session` each have their own `id` / `create` — they are mutually exclusive. * When passing an existing `session["id"]`, **do not** also pass `agent`. * `session["create"]` must include `environment_id` explicitly. * The Cloud runtime rejects local-CLI-only top-level options: `model`, `agent`, `mcp_servers`, `settings`, `hooks`, `plugins`, `permission_mode`, etc. Passing any of them throws synchronously. * The Cloud runtime does not support `QoderSDKClient.set_model()`, `reload_plugins()`, MCP OAuth, etc. Only `async for` consumption and `close()` are guaranteed.
## Error codes | Exception class | When it triggers | | -------------------------------- | -------------------------------------------------------------- | | `CloudAgentUnsupportedAuthError` | Non-PAT auth was used (e.g. `qodercli_auth()` / `job_token()`) | | `CloudAgentApiError` | Cloud OpenAPI returned non-2xx, or the SSE channel failed |