qwen-code/packages/sdk-python
jinye 2c93fd670c
refactor: extract shared release helper utilities (#3834)
Move four duplicated utility functions (getArgs, readJson,
validateVersion, isExpectedMissingGitHubRelease) from the three
get-release-version.js scripts into a shared module at
scripts/lib/release-helpers.js so that changes only need to happen
in one place.

Also fixes a pre-existing bug in getArgs where argument values
containing '=' were silently truncated (e.g. --msg=a=b produced
{msg:'a'} instead of {msg:'a=b'}).

Closes #3795

🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code)

Co-authored-by: jinye.djy <jinye.djy@alibaba-inc.com>
2026-05-05 10:15:17 +08:00
..
scripts refactor: extract shared release helper utilities (#3834) 2026-05-05 10:15:17 +08:00
src/qwen_code_sdk feat(SDK) Add Python SDK implementation for #3010 (#3494) 2026-04-25 07:02:58 +08:00
tests feat(SDK) Add Python SDK implementation for #3010 (#3494) 2026-04-25 07:02:58 +08:00
pyproject.toml feat(SDK) Add Python SDK implementation for #3010 (#3494) 2026-04-25 07:02:58 +08:00
README.md feat(SDK) Add Python SDK implementation for #3010 (#3494) 2026-04-25 07:02:58 +08:00

qwen-code-sdk

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

Installation

pip install 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.

Quick Start

import asyncio

from qwen_code_sdk import is_sdk_result_message, query


async def main() -> None:
    result = query(
        "List the top-level packages in this repository.",
        {
            "cwd": "/path/to/project",
            "path_to_qwen_executable": "qwen",
        },
    )

    async for message in result:
        if is_sdk_result_message(message):
            print(message["result"])


asyncio.run(main())

Sync API

from qwen_code_sdk import query_sync


with query_sync(
    "Say hello",
    {
        "path_to_qwen_executable": "qwen",
    },
) as result:
    for message in result:
        print(message)

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.

Permission Callback

from qwen_code_sdk import query


async def can_use_tool(tool_name, tool_input, context):
    if tool_name == "write_file":
        return {"behavior": "deny", "message": "Writes disabled in this app"}
    return {"behavior": "allow", "updatedInput": tool_input}


result = query(
    "Create hello.txt",
    {
        "path_to_qwen_executable": "qwen",
        "can_use_tool": can_use_tool,
    },
)

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

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.

Errors

  • 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

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.