vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-05-19 07:54:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

doc[sdk-python] Expand Python SDK usage documentation (#3995)

Browse source 
* docs(sdk-python): expand usage examples

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* 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 <qwen-coder@alibabacloud.com>

---------

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
jinye 2026-05-12 15:27:00 +08:00 • committed by GitHub
parent 70eecdbdf4
commit 826f9fd126
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 710 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

409
docs/developers/sdk-python.md
View file

@ -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:

357
packages/sdk-python/README.md
View file

@ -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: