qwen-code/docs/developers/sdk-python.md

Python SDK

qwen-code-sdk

qwen-code-sdk is an experimental Python SDK for Qwen Code. v1 targets the existing stream-json CLI protocol and keeps the transport surface small and testable.

Scope

• Package name: qwen-code-sdk
• Import path: qwen_code_sdk
• Runtime requirement: Python >=3.10
• CLI dependency: external qwen executable is required in v1
• Transport scope: process transport only
• Not included in v1: ACP transport, SDK-embedded MCP servers

Install

pip install qwen-code-sdk

For preview releases:

pip install --pre qwen-code-sdk

If qwen is not on PATH, pass path_to_qwen_executable explicitly.

Before writing SDK code, make sure the CLI works in the same shell:

qwen --version

Quick Start

import asyncio

from qwen_code_sdk import (
    is_sdk_assistant_message,
    is_sdk_result_message,
    query,
)


def extract_text(message):
    content = message.get("message", {}).get("content", [])
    if not isinstance(content, list):
        return repr(content)
    texts = [
        block.get("text", "")
        for block in content
        if isinstance(block, dict) and block.get("type") == "text"
    ]
    return "".join(texts) if texts else "[no text content]"


def print_result(message):
    if message.get("is_error"):
        error = message.get("error") or {}
        print(f"Error: {error.get('message', 'Unknown error')}")
        return
    print(message.get("result", ""))


async def main() -> None:
    async with query(
        "Explain the repository structure.",
        {
            "cwd": "/path/to/project",
            "path_to_qwen_executable": "qwen",
        },
    ) as result:
        async for message in result:
            if is_sdk_assistant_message(message):
                print(extract_text(message))
            elif is_sdk_result_message(message):
                print_result(message)


asyncio.run(main())

asyncio.run() is appropriate for standalone scripts. If your application already runs an event loop, such as Jupyter, FastAPI, or pytest-asyncio, call await main() instead.

Sync Usage

Use query_sync when your host application is not async:

from qwen_code_sdk import is_sdk_result_message, query_sync


with query_sync(
    "Summarize this repository in one paragraph.",
    {
        "cwd": "/path/to/project",
        "path_to_qwen_executable": "qwen",
    },
) as result:
    for message in result:
        if is_sdk_result_message(message):
            if message.get("is_error"):
                error = message.get("error") or {}
                print(f"Error: {error.get('message', 'Unknown error')}")
            else:
                print(message.get("result", ""))

API Surface

Top-level entry points

• query(prompt, options=None) -> Query
• query_sync(prompt, options=None) -> SyncQuery

prompt supports either:

• str for single-turn requests
• AsyncIterable[SDKUserMessage] for multi-turn streams

Query

• Async iterable over SDK messages
• close()
• interrupt()
• set_model(model)
• set_permission_mode(mode)
• supported_commands()
• mcp_server_status()
• get_session_id()
• is_closed()

QueryOptions

Option	Type / values	Description
cwd	str	Working directory for the CLI process.
model	str	Model override for this SDK session.
path_to_qwen_executable	str	qwen, an explicit binary path, or a .js CLI bundle.
permission_mode	default, plan, auto-edit, yolo	Tool execution approval mode. yolo auto-approves all tools; use it only in trusted or sandboxed environments.
can_use_tool	async callback	Custom permission callback for tool requests.
env	dict[str, str]	Extra environment variables passed to the CLI process.
system_prompt	str	Override the system prompt.
append_system_prompt	str	Append extra instructions to the system prompt.
debug	bool	Forward CLI stderr to stderr when no stderr hook exists.
max_session_turns	int	Maximum turns before the CLI ends the session.
core_tools	list[str]	Restrict the available tool set.
exclude_tools	list[str]	Exclude matching tools.
allowed_tools	list[str]	Allow matching tools without callback approval.
auth_type	openai, anthropic, qwen-oauth, gemini, vertex-ai	Authentication mode passed to the CLI.
include_partial_messages	bool	Emit partial assistant stream events.
resume	UUID string	Resume a known session id.
continue_session	bool	Continue the latest CLI session.
session_id	UUID string	Start or correlate a session with a known id.
timeout	mapping	Timeouts in seconds.
stderr	callable	Receives CLI stderr lines.

Use only one of resume, continue_session, or session_id in a request. The SDK raises ValidationError if these session options are combined.

Unsupported in v1:

• mcp_servers

Common Configuration

options = {
    "cwd": "/path/to/project",
    "path_to_qwen_executable": "qwen",
    "model": "qwen-plus",
    "permission_mode": "plan",
    "max_session_turns": 1,
    "env": {
        "OPENAI_MODEL": "qwen-plus",
    },
    "timeout": {
        "control_request": 60,
        "can_use_tool": 60,
        "stream_close": 60,
    },
}

Timeout values are seconds. env is merged on top of the parent process environment, so you only need to pass variables that should differ for this SDK session. Set secrets such as OPENAI_API_KEY in the parent environment or a secrets manager rather than hardcoding them in source.

Permission Handling

When the CLI emits a can_use_tool control request, the SDK routes it through can_use_tool(tool_name, tool_input, context).

• Default behavior: deny
• Default timeout: 60 seconds, configurable with timeout.can_use_tool
• Timeout fallback: deny
• Callback exceptions: converted to deny with an error message
• Callback context: cancel_event, suggestions, and blocked_path
• Callback contract: can_use_tool must be async with 3 positional arguments; stderr must accept 1 positional string argument

Example:

import asyncio
from pathlib import Path

from qwen_code_sdk import is_sdk_result_message, query

PROJECT_ROOT = Path("/path/to/project").resolve()


def project_path(tool_name, tool_input):
    key = "path" if tool_name == "list_directory" else "file_path"
    raw_path = tool_input.get(key)
    if not isinstance(raw_path, str) or not raw_path:
        return None

    resolved = (PROJECT_ROOT / raw_path).resolve()
    try:
        resolved.relative_to(PROJECT_ROOT)
    except ValueError:
        return None
    return resolved


async def can_use_tool(tool_name, tool_input, context):
    if tool_name in {"read_file", "list_directory", "write_file"}:
        resolved = project_path(tool_name, tool_input)
        if resolved is None:
            return {
                "behavior": "deny",
                "message": "Only project-local paths are allowed",
            }

        if tool_name == "write_file" and resolved.suffix != ".md":
            return {"behavior": "deny", "message": "Only .md files can be written"}

        return {"behavior": "allow", "updatedInput": tool_input}

    return {
        "behavior": "deny",
        "message": f"{tool_name} is not allowed by this application",
    }


async def main():
    async with query(
        "Update README.md with a short summary.",
        {
            "cwd": str(PROJECT_ROOT),
            "path_to_qwen_executable": "qwen",
            "can_use_tool": can_use_tool,
        },
    ) as result:
        async for message in result:
            if is_sdk_result_message(message):
                if message.get("is_error"):
                    error = message.get("error") or {}
                    print(f"Error: {error.get('message', 'Unknown error')}")
                else:
                    print(message.get("result", ""))


asyncio.run(main())

If you do not pass can_use_tool, the SDK denies permission requests by default.

Multi-Turn Sessions

For multi-turn sessions, pass an async iterable of SDKUserMessage objects:

import asyncio

from qwen_code_sdk import SDKUserMessage, is_sdk_result_message, query

SESSION_ID = "123e4567-e89b-12d3-a456-426614174000"


async def prompts():
    first: SDKUserMessage = {
        "type": "user",
        "session_id": SESSION_ID,
        "message": {
            "role": "user",
            "content": "Create a concise project summary.",
        },
        "parent_tool_use_id": None,
    }
    yield first

    second: SDKUserMessage = {
        "type": "user",
        "session_id": SESSION_ID,
        "message": {
            "role": "user",
            "content": "Also list the test files.",
        },
        "parent_tool_use_id": None,
    }
    yield second


async def main():
    async with query(
        prompts(),
        {
            "cwd": "/path/to/project",
            "path_to_qwen_executable": "qwen",
            "session_id": SESSION_ID,
        },
    ) as result:
        async for message in result:
            if is_sdk_result_message(message):
                if message.get("is_error"):
                    error = message.get("error") or {}
                    print(f"Error: {error.get('message', 'Unknown error')}")
                else:
                    print(message.get("result", ""))


asyncio.run(main())

All messages in the async iterable must be known upfront. The SDK sends them sequentially to the CLI but cannot feed a prior response back into the generator. If you need conversational turn-taking, manage each turn as a separate query() call.

Runtime Controls

The returned Query object can control the running CLI process:

import asyncio

from qwen_code_sdk import is_sdk_result_message, query


async def main():
    async with query(
        "Inspect this repository and explain the test layout.",
        {
            "cwd": "/path/to/project",
            "path_to_qwen_executable": "qwen",
        },
    ) as result:
        commands = await result.supported_commands()
        print(commands)

        await result.set_permission_mode("plan")
        await result.set_model("qwen-plus")

        async for message in result:
            if is_sdk_result_message(message):
                if message.get("is_error"):
                    error = message.get("error") or {}
                    print(f"Error: {error.get('message', 'Unknown error')}")
                else:
                    print(message.get("result", ""))


asyncio.run(main())

Use interrupt() to cancel the current operation, close() to clean up the underlying process, and get_session_id() to persist a session id for later.

Session Resume

import asyncio

from qwen_code_sdk import is_sdk_result_message, query


async def main():
    # Resume a known session by its id.
    async with query(
        "Continue from this session.",
        {
            "path_to_qwen_executable": "qwen",
            "resume": "123e4567-e89b-12d3-a456-426614174000",
        },
    ) as known:
        async for message in known:
            if is_sdk_result_message(message):
                if message.get("is_error"):
                    error = message.get("error") or {}
                    print(f"Error: {error.get('message', 'Unknown error')}")
                else:
                    print(message.get("result", ""))


asyncio.run(main())

To continue the latest session instead:

import asyncio

from qwen_code_sdk import is_sdk_result_message, query


async def main():
    async with query(
        "Continue the latest session.",
        {
            "path_to_qwen_executable": "qwen",
            "continue_session": True,
        },
    ) as latest:
        async for message in latest:
            if is_sdk_result_message(message):
                if message.get("is_error"):
                    error = message.get("error") or {}
                    print(f"Error: {error.get('message', 'Unknown error')}")
                else:
                    print(message.get("result", ""))


asyncio.run(main())

resume is useful when your application stores session ids. continue_session delegates the selection of the latest session to the CLI.

Error Model

• ValidationError: invalid options, invalid UUIDs, unsupported combinations
• ControlRequestTimeoutError: initialize, interrupt, or other control request timed out
• ProcessExitError: CLI exited non-zero
• AbortError: control request or session was cancelled

from qwen_code_sdk import (
    ProcessExitError,
    ValidationError,
    is_sdk_result_message,
    query_sync,
)

try:
    with query_sync("Say hello", {"path_to_qwen_executable": "qwen"}) as result:
        for message in result:
            if is_sdk_result_message(message):
                if message.get("is_error"):
                    error = message.get("error") or {}
                    print(f"Error: {error.get('message', 'Unknown error')}")
                else:
                    print(message.get("result", ""))
except ValidationError as exc:
    print(f"Invalid SDK options: {exc}")
except ProcessExitError as exc:
    print(f"qwen exited with {exc.exit_code}: {exc}")

Troubleshooting

If the SDK cannot start the CLI:

• Verify qwen --version works in the target environment
• Pass path_to_qwen_executable if your shell uses nvm, pyenv, or other non-standard PATH setup
• Use debug=True or stderr=print to surface CLI stderr while debugging

If session control calls time out:

• Check that the target qwen version supports --input-format stream-json
• Increase timeout.control_request
• Verify that no wrapper script is swallowing stdout/stderr

Repository Integration

Repository-level helper commands:

• npm run test:sdk:python
• npm run lint:sdk:python
• npm run typecheck:sdk:python
• npm run smoke:sdk:python -- --qwen qwen

Real E2E Smoke

For a real runtime check (actual qwen process + real model call), run from the repository root. The npm helper uses python3, so ensure it resolves to a Python >=3.10 interpreter:

npm run smoke:sdk:python -- --qwen qwen

This script runs:

• async single-turn query
• async control flow (supported_commands, permission mode updates)
• sync query_sync query

It prints JSON and returns non-zero on failure.