Skip to main content
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; see SDK 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.
export QODER_PERSONAL_ACCESS_TOKEN="<your-pat>"
export QODER_CLOUD_AGENT_ENVIRONMENT_ID="<your-env-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: 
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).

​
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: 
"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: 
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. 
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: 
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: 
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):
FieldDescription
eventCloud event name (e.g. user.message, agent.message, session.status_idle)
idEvent ID in the SSE stream; usable as a replay anchor
dataCloud event payload (includes turn_id and other fields)
uuidSDK-generated unique ID for deduplication
session_idCloud 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

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

FieldDescription
subtype"success" or an error subtype
is_errorBoolean; whether the turn ended abnormally
session_idCloud session ID (backfilled by the SDK on the create branch)
resultAgent’s text reply for the turn (multiple text blocks are concatenated)
usage / model_usage / total_cost_usdBackfilled 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 classWhen it triggers
CloudAgentUnsupportedAuthErrorNon-PAT auth was used (e.g. qodercli_auth() / job_token())
CloudAgentApiErrorCloud OpenAPI returned non-2xx, or the SSE channel failed