# 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 ```bash pip install qwen-code-sdk ``` For preview releases: ```bash 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: ```bash qwen --version ``` ## Quick Start ```python 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: ```python 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 ```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, }, } ``` 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: ```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 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: ```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 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: ```python 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 ```python 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: ```python 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 ```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}") ``` ## 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: ```bash 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.