From 826f9fd12609f27cd023c4a61cd5dd0f3cb87d2d Mon Sep 17 00:00:00 2001 From: jinye Date: Tue, 12 May 2026 15:27:00 +0800 Subject: [PATCH] doc[sdk-python] Expand Python SDK usage documentation (#3995) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * docs(sdk-python): expand usage examples Co-authored-by: Qwen-Coder * fix(docs): correct file_path key and update session resume examples * fix(docs): add is_error handling and async iteration to SDK examples - Session Resume examples now check is_error before printing result, consistent with the print_result helper in Quick Start - Permission Callback examples now wrap query() in async def main() with async for iteration, so the CLI process actually starts 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * docs(sdk-python): address review feedback Co-authored-by: Qwen-Coder --------- Co-authored-by: Qwen-Coder --- docs/developers/sdk-python.md | 409 +++++++++++++++++++++++++++++++--- packages/sdk-python/README.md | 357 +++++++++++++++++++++++++++-- 2 files changed, 710 insertions(+), 56 deletions(-) diff --git a/docs/developers/sdk-python.md b/docs/developers/sdk-python.md index 7ef2c0ab6..c39f5a9d2 100644 --- a/docs/developers/sdk-python.md +++ b/docs/developers/sdk-python.md @@ -21,33 +21,98 @@ testable. 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_result_message, query +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: - result = query( + async with query( "Explain the repository structure.", { "cwd": "/path/to/project", "path_to_qwen_executable": "qwen", }, - ) - - async for message in result: - if is_sdk_result_message(message): - print(message["result"]) + ) 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 @@ -74,35 +139,60 @@ asyncio.run(main()) ### `QueryOptions` -Supported options in v1: +| 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: -- `cwd` -- `model` -- `path_to_qwen_executable` -- `permission_mode` -- `can_use_tool` -- `env` -- `system_prompt` -- `append_system_prompt` -- `debug` -- `max_session_turns` -- `core_tools` -- `exclude_tools` -- `allowed_tools` -- `auth_type` -- `include_partial_messages` -- `resume` -- `continue_session` -- `session_id` -- `timeout` - `mcp_servers` -- `stderr` -Session argument priority is fixed as: +### Common Configuration -1. `resume` -2. `continue_session` -3. `session_id` +```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 @@ -110,13 +200,243 @@ 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 +- 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 @@ -125,6 +445,29 @@ When the CLI emits a `can_use_tool` control request, the SDK routes it through - `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: diff --git a/packages/sdk-python/README.md b/packages/sdk-python/README.md index 67cf3ae34..3f324a421 100644 --- a/packages/sdk-python/README.md +++ b/packages/sdk-python/README.md @@ -9,6 +9,12 @@ Experimental Python SDK for programmatic access to Qwen Code through the pip install qwen-code-sdk ``` +For preview releases, enable pre-release resolution: + +```bash +pip install --pre qwen-code-sdk +``` + ## Requirements - Python `>=3.10` @@ -17,45 +23,86 @@ pip install qwen-code-sdk You can also point the SDK at an explicit CLI binary or script with `path_to_qwen_executable`. +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_result_message, query +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: - result = query( + async with 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"]) + ) 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 query_sync +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: - print(message) + 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 @@ -68,42 +115,306 @@ with query_sync( `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`, or `yolo`; `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`, `exclude_tools`, `allowed_tools`: constrain tool availability +- `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 -from qwen_code_sdk import query +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 == "write_file": - return {"behavior": "deny", "message": "Writes disabled in this app"} - return {"behavior": "allow", "updatedInput": tool_input} + 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", + } -result = query( - "Create hello.txt", - { - "path_to_qwen_executable": "qwen", - "can_use_tool": can_use_tool, - }, -) +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 60 seconds, the SDK -auto-denies the tool request. +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`. -## Errors +## 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: