# qwen-code-sdk

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

## Installation

```bash
pip install qwen-code-sdk
```

For preview releases, enable pre-release resolution:

```bash
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:

```bash
qwen --version
```

## Quick Start

```python
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

```python
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

```python
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:

```python
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

```python
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").

```python
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:

```python
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

```python
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:

```python
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

```python
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](../../docs/developers/sdk-python.md) for more
detail.