Skip to content

README.md

Latest commit

 

History

History
475 lines (367 loc) · 14 KB

README.md

File metadata and controls

475 lines (367 loc) · 14 KB
Raw

qwen-code-sdk

Experimental Python SDK for programmatic access to Qwen Code through the stream-json protocol.

Installation

pip install qwen-code-sdk

For preview releases, enable pre-release resolution:

pip install --pre qwen-code-sdk

Requirements

  • Python >=3.10
  • External qwen CLI installed and available in PATH

You can also point the SDK at an explicit CLI binary or script with path_to_qwen_executable.

On Windows, command names are resolved only from the effective child PATH, including case-insensitive overrides in QueryOptions.env. Supported command extensions are .com, .exe, .bat, and .cmd; extensionless npm shell shims and unrelated PATHEXT associations are ignored. The current directory is searched only when explicitly included in PATH. Explicit Python or JavaScript file paths continue to use their corresponding runtime.

Closing a running Windows command shim uses the system taskkill /T /F command to stop both the shim and its CLI descendants before the parent process exits. Exit and stderr cleanup waits are bounded so an inherited pipe cannot stall shutdown.

Before using the SDK, verify that the CLI works in the same environment:

qwen --version

Quick Start

import asyncio

from qwen_code_sdk import (
    is_sdk_assistant_message,
    is_sdk_result_message,
    query,
)


def text_from_message(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(
        "List the top-level packages in this repository.",
        {
            "cwd": "/path/to/project",
            "path_to_qwen_executable": "qwen",
        },
    ) as result:
        async for message in result:
            if is_sdk_assistant_message(message):
                print(text_from_message(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 API

from qwen_code_sdk import is_sdk_result_message, query_sync


with query_sync(
    "Say hello",
    {
        "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", ""))

Main APIs

  • query(prompt, options=None) -> Query
  • query_sync(prompt, options=None) -> SyncQuery
  • Query.close(), interrupt(), set_model(), set_permission_mode()
  • Query.supported_commands(), mcp_server_status(), get_session_id()

prompt accepts either a single str or an AsyncIterable[SDKUserMessage] for multi-turn sessions.

Common Options

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,
    },
}

Common fields:

  • cwd: working directory used by the CLI
  • path_to_qwen_executable: qwen, an absolute binary path, or a .js CLI bundle
  • model: model override for this session
  • permission_mode: one of default, plan, auto-edit, auto, or yolo; auto lets an LLM classifier approve tool calls; yolo auto-approves all tools, so use it only in trusted or sandboxed environments
  • env: extra environment variables passed to the CLI process
  • system_prompt / append_system_prompt: override or extend the system prompt
  • core_tools: a non-empty allowlist of core tools. An omitted or empty list is treated as unspecified; core_tools=[] does not disable tools. MCP and other non-core tools have separate registration and permission rules.
  • exclude_tools: exclude named tools explicitly.
  • allowed_tools: bypass approval for matching tools; this is not a tool-registration allowlist.
  • timeout: seconds for control requests, permission callbacks, and stream close waits

env is merged on top of the parent process environment. Set secrets such as OPENAI_API_KEY in the parent environment or a secrets manager rather than hardcoding them in source.

Multi-Turn Sessions

For multi-turn use cases, pass an async iterable of SDKUserMessage objects. Use a stable UUID for session_id when you want to correlate messages:

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 short 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.

Permission Callback

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 one paragraph 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())

The callback defaults to deny. If it does not return within timeout.can_use_tool seconds, the SDK auto-denies the tool request. The default timeout is 60 seconds.

The context argument includes cancel_event, suggestions, and blocked_path when the CLI provides a path-specific permission target. can_use_tool must be an async def callback accepting (tool_name, tool_input, context). stderr must accept a single str.

Handling ask_user_question

When the model needs a decision from the user it calls the built-in ask_user_question tool. This flows through the same can_use_tool callback: tool_input carries a questions list, and you return the collected answers via updatedInput["answers"]. answers is a dict keyed by the question's index (as a string), where each value is the label of the chosen option (or free-form text when the user picks "Other").

async def can_use_tool(tool_name, tool_input, context):
    if tool_name == "ask_user_question":
        questions = tool_input["questions"]

        # Present the questions to the user however your app sees fit,
        # then build an index-keyed map of their answers.
        answers = {}
        for index, question in enumerate(questions):
            answers[str(index)] = await prompt_user_to_choose(question)

        # Return the answers through updatedInput["answers"] — the CLI
        # forwards them to the tool so the model receives the decisions.
        return {
            "behavior": "allow",
            "updatedInput": {**tool_input, "answers": answers},
        }

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

If you return allow without any answers, the tool reports that no answer was provided; return deny to signal the user declined.

Runtime Controls

Control methods can be called while a session is active:

import asyncio

from qwen_code_sdk import is_sdk_result_message, query


async def main():
    async with query(
        "Inspect this project and wait for my next instruction.",
        {
            "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 CLI operation and close() to clean up the underlying process.

Resuming Sessions

import asyncio

from qwen_code_sdk import is_sdk_result_message, query


async def main():
    # Resume a known session.
    async with query(
        "Continue from the previous state.",
        {
            "path_to_qwen_executable": "qwen",
            "resume": "123e4567-e89b-12d3-a456-426614174000",
        },
    ) 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())

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 last 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())

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

Error Handling

  • ValidationError: invalid query options or malformed session identifiers
  • ControlRequestTimeoutError: CLI control operation exceeded timeout
  • ProcessExitError: qwen exited with a non-zero code
  • AbortError: query or control request 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}")

Current Scope

0.1.x is intentionally narrow:

  • Uses external qwen CLI via process transport
  • Targets stream-json parity with the TypeScript SDK core flow
  • Does not yet implement ACP transport
  • Does not yet embed MCP servers inside the SDK process

See developer documentation for more detail.