basant307/AI_Governance_Project
045
1# Python SDK2 3## `qwen-code-sdk`4 5`qwen-code-sdk` is an experimental Python SDK for Qwen Code. v1 targets the6existing `stream-json` CLI protocol and keeps the transport surface small and7testable.8 9## Scope10 11- Package name: `qwen-code-sdk`12- Import path: `qwen_code_sdk`13- Runtime requirement: Python `>=3.10`14- CLI dependency: external `qwen` executable is required in v115- Transport scope: process transport only16- Not included in v1: ACP transport, SDK-embedded MCP servers17 18## Install19 20```bash21pip install qwen-code-sdk22```23 24For preview releases:25 26```bash27pip install --pre qwen-code-sdk28```29 30If `qwen` is not on `PATH`, pass `path_to_qwen_executable` explicitly.31 32Before writing SDK code, make sure the CLI works in the same shell:33 34```bash35qwen --version36```37 38## Quick Start39 40```python41import asyncio42 43from qwen_code_sdk import (44 is_sdk_assistant_message,45 is_sdk_result_message,46 query,47)48 49 50def extract_text(message):51 content = message.get("message", {}).get("content", [])52 if not isinstance(content, list):53 return repr(content)54 texts = [55 block.get("text", "")56 for block in content57 if isinstance(block, dict) and block.get("type") == "text"58 ]59 return "".join(texts) if texts else "[no text content]"60 61 62def print_result(message):63 if message.get("is_error"):64 error = message.get("error") or {}65 print(f"Error: {error.get('message', 'Unknown error')}")66 return67 print(message.get("result", ""))68 69 70async def main() -> None:71 async with query(72 "Explain the repository structure.",73 {74 "cwd": "/path/to/project",75 "path_to_qwen_executable": "qwen",76 },77 ) as result:78 async for message in result:79 if is_sdk_assistant_message(message):80 print(extract_text(message))81 elif is_sdk_result_message(message):82 print_result(message)83 84 85asyncio.run(main())86```87 88`asyncio.run()` is appropriate for standalone scripts. If your application89already runs an event loop, such as Jupyter, FastAPI, or pytest-asyncio, call90`await main()` instead.91 92## Sync Usage93 94Use `query_sync` when your host application is not async:95 96```python97from qwen_code_sdk import is_sdk_result_message, query_sync98 99 100with query_sync(101 "Summarize this repository in one paragraph.",102 {103 "cwd": "/path/to/project",104 "path_to_qwen_executable": "qwen",105 },106) as result:107 for message in result:108 if is_sdk_result_message(message):109 if message.get("is_error"):110 error = message.get("error") or {}111 print(f"Error: {error.get('message', 'Unknown error')}")112 else:113 print(message.get("result", ""))114```115 116## API Surface117 118### Top-level entry points119 120- `query(prompt, options=None) -> Query`121- `query_sync(prompt, options=None) -> SyncQuery`122 123`prompt` supports either:124 125- `str` for single-turn requests126- `AsyncIterable[SDKUserMessage]` for multi-turn streams127 128### `Query`129 130- Async iterable over SDK messages131- `close()`132- `interrupt()`133- `set_model(model)`134- `set_permission_mode(mode)`135- `supported_commands()`136- `mcp_server_status()`137- `get_session_id()`138- `is_closed()`139 140### `QueryOptions`141 142| Option | Type / values | Description |143| -------------------------- | ---------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------- |144| `cwd` | `str` | Working directory for the CLI process. |145| `model` | `str` | Model override for this SDK session. |146| `path_to_qwen_executable` | `str` | `qwen`, an explicit binary path, or a `.js` CLI bundle. |147| `permission_mode` | `default`, `plan`, `auto-edit`, `yolo` | Tool execution approval mode. `yolo` auto-approves all tools; use it only in trusted or sandboxed environments. |148| `can_use_tool` | async callback | Custom permission callback for tool requests. |149| `env` | `dict[str, str]` | Extra environment variables passed to the CLI process. |150| `system_prompt` | `str` | Override the system prompt. |151| `append_system_prompt` | `str` | Append extra instructions to the system prompt. |152| `debug` | `bool` | Forward CLI stderr to stderr when no `stderr` hook exists. |153| `max_session_turns` | `int` | Maximum turns before the CLI ends the session. |154| `core_tools` | `list[str]` | Restrict the available tool set. |155| `exclude_tools` | `list[str]` | Exclude matching tools. |156| `allowed_tools` | `list[str]` | Allow matching tools without callback approval. |157| `auth_type` | `openai`, `anthropic`, `qwen-oauth`, `gemini`, `vertex-ai` | Authentication mode passed to the CLI. |158| `include_partial_messages` | `bool` | Emit partial assistant stream events. |159| `resume` | UUID string | Resume a known session id. |160| `continue_session` | `bool` | Continue the latest CLI session. |161| `session_id` | UUID string | Start or correlate a session with a known id. |162| `timeout` | mapping | Timeouts in seconds. |163| `stderr` | callable | Receives CLI stderr lines. |164 165Use only one of `resume`, `continue_session`, or `session_id` in a request. The166SDK raises `ValidationError` if these session options are combined.167 168Unsupported in v1:169 170- `mcp_servers`171 172### Common Configuration173 174```python175options = {176 "cwd": "/path/to/project",177 "path_to_qwen_executable": "qwen",178 "model": "qwen-plus",179 "permission_mode": "plan",180 "max_session_turns": 1,181 "env": {182 "OPENAI_MODEL": "qwen-plus",183 },184 "timeout": {185 "control_request": 60,186 "can_use_tool": 60,187 "stream_close": 60,188 },189}190```191 192Timeout values are seconds. `env` is merged on top of the parent process193environment, so you only need to pass variables that should differ for this SDK194session. Set secrets such as `OPENAI_API_KEY` in the parent environment or a195secrets manager rather than hardcoding them in source.196 197## Permission Handling198 199When the CLI emits a `can_use_tool` control request, the SDK routes it through200`can_use_tool(tool_name, tool_input, context)`.201 202- Default behavior: deny203- Default timeout: 60 seconds, configurable with `timeout.can_use_tool`204- Timeout fallback: deny205- Callback exceptions: converted to deny with an error message206- Callback context: `cancel_event`, `suggestions`, and `blocked_path`207- Callback contract: `can_use_tool` must be async with 3 positional arguments;208 `stderr` must accept 1 positional string argument209 210Example:211 212```python213import asyncio214from pathlib import Path215 216from qwen_code_sdk import is_sdk_result_message, query217 218PROJECT_ROOT = Path("/path/to/project").resolve()219 220 221def project_path(tool_name, tool_input):222 key = "path" if tool_name == "list_directory" else "file_path"223 raw_path = tool_input.get(key)224 if not isinstance(raw_path, str) or not raw_path:225 return None226 227 resolved = (PROJECT_ROOT / raw_path).resolve()228 try:229 resolved.relative_to(PROJECT_ROOT)230 except ValueError:231 return None232 return resolved233 234 235async def can_use_tool(tool_name, tool_input, context):236 if tool_name in {"read_file", "list_directory", "write_file"}:237 resolved = project_path(tool_name, tool_input)238 if resolved is None:239 return {240 "behavior": "deny",241 "message": "Only project-local paths are allowed",242 }243 244 if tool_name == "write_file" and resolved.suffix != ".md":245 return {"behavior": "deny", "message": "Only .md files can be written"}246 247 return {"behavior": "allow", "updatedInput": tool_input}248 249 return {250 "behavior": "deny",251 "message": f"{tool_name} is not allowed by this application",252 }253 254 255async def main():256 async with query(257 "Update README.md with a short summary.",258 {259 "cwd": str(PROJECT_ROOT),260 "path_to_qwen_executable": "qwen",261 "can_use_tool": can_use_tool,262 },263 ) as result:264 async for message in result:265 if is_sdk_result_message(message):266 if message.get("is_error"):267 error = message.get("error") or {}268 print(f"Error: {error.get('message', 'Unknown error')}")269 else:270 print(message.get("result", ""))271 272 273asyncio.run(main())274```275 276If you do not pass `can_use_tool`, the SDK denies permission requests by277default.278 279## Multi-Turn Sessions280 281For multi-turn sessions, pass an async iterable of `SDKUserMessage` objects:282 283```python284import asyncio285 286from qwen_code_sdk import SDKUserMessage, is_sdk_result_message, query287 288SESSION_ID = "123e4567-e89b-12d3-a456-426614174000"289 290 291async def prompts():292 first: SDKUserMessage = {293 "type": "user",294 "session_id": SESSION_ID,295 "message": {296 "role": "user",297 "content": "Create a concise project summary.",298 },299 "parent_tool_use_id": None,300 }301 yield first302 303 second: SDKUserMessage = {304 "type": "user",305 "session_id": SESSION_ID,306 "message": {307 "role": "user",308 "content": "Also list the test files.",309 },310 "parent_tool_use_id": None,311 }312 yield second313 314 315async def main():316 async with query(317 prompts(),318 {319 "cwd": "/path/to/project",320 "path_to_qwen_executable": "qwen",321 "session_id": SESSION_ID,322 },323 ) as result:324 async for message in result:325 if is_sdk_result_message(message):326 if message.get("is_error"):327 error = message.get("error") or {}328 print(f"Error: {error.get('message', 'Unknown error')}")329 else:330 print(message.get("result", ""))331 332 333asyncio.run(main())334```335 336All messages in the async iterable must be known upfront. The SDK sends them337sequentially to the CLI but cannot feed a prior response back into the generator.338If you need conversational turn-taking, manage each turn as a separate `query()`339call.340 341## Runtime Controls342 343The returned `Query` object can control the running CLI process:344 345```python346import asyncio347 348from qwen_code_sdk import is_sdk_result_message, query349 350 351async def main():352 async with query(353 "Inspect this repository and explain the test layout.",354 {355 "cwd": "/path/to/project",356 "path_to_qwen_executable": "qwen",357 },358 ) as result:359 commands = await result.supported_commands()360 print(commands)361 362 await result.set_permission_mode("plan")363 await result.set_model("qwen-plus")364 365 async for message in result:366 if is_sdk_result_message(message):367 if message.get("is_error"):368 error = message.get("error") or {}369 print(f"Error: {error.get('message', 'Unknown error')}")370 else:371 print(message.get("result", ""))372 373 374asyncio.run(main())375```376 377Use `interrupt()` to cancel the current operation, `close()` to clean up the378underlying process, and `get_session_id()` to persist a session id for later.379 380## Session Resume381 382```python383import asyncio384 385from qwen_code_sdk import is_sdk_result_message, query386 387 388async def main():389 # Resume a known session by its id.390 async with query(391 "Continue from this session.",392 {393 "path_to_qwen_executable": "qwen",394 "resume": "123e4567-e89b-12d3-a456-426614174000",395 },396 ) as known:397 async for message in known:398 if is_sdk_result_message(message):399 if message.get("is_error"):400 error = message.get("error") or {}401 print(f"Error: {error.get('message', 'Unknown error')}")402 else:403 print(message.get("result", ""))404 405 406asyncio.run(main())407```408 409To continue the latest session instead:410 411```python412import asyncio413 414from qwen_code_sdk import is_sdk_result_message, query415 416 417async def main():418 async with query(419 "Continue the latest session.",420 {421 "path_to_qwen_executable": "qwen",422 "continue_session": True,423 },424 ) as latest:425 async for message in latest:426 if is_sdk_result_message(message):427 if message.get("is_error"):428 error = message.get("error") or {}429 print(f"Error: {error.get('message', 'Unknown error')}")430 else:431 print(message.get("result", ""))432 433 434asyncio.run(main())435```436 437`resume` is useful when your application stores session ids. `continue_session`438delegates the selection of the latest session to the CLI.439 440## Error Model441 442- `ValidationError`: invalid options, invalid UUIDs, unsupported combinations443- `ControlRequestTimeoutError`: initialize, interrupt, or other control request444 timed out445- `ProcessExitError`: CLI exited non-zero446- `AbortError`: control request or session was cancelled447 448```python449from qwen_code_sdk import (450 ProcessExitError,451 ValidationError,452 is_sdk_result_message,453 query_sync,454)455 456try:457 with query_sync("Say hello", {"path_to_qwen_executable": "qwen"}) as result:458 for message in result:459 if is_sdk_result_message(message):460 if message.get("is_error"):461 error = message.get("error") or {}462 print(f"Error: {error.get('message', 'Unknown error')}")463 else:464 print(message.get("result", ""))465except ValidationError as exc:466 print(f"Invalid SDK options: {exc}")467except ProcessExitError as exc:468 print(f"qwen exited with {exc.exit_code}: {exc}")469```470 471## Troubleshooting472 473If the SDK cannot start the CLI:474 475- Verify `qwen --version` works in the target environment476- Pass `path_to_qwen_executable` if your shell uses `nvm`, `pyenv`, or other477 non-standard PATH setup478- Use `debug=True` or `stderr=print` to surface CLI stderr while debugging479 480If session control calls time out:481 482- Check that the target `qwen` version supports `--input-format stream-json`483- Increase `timeout.control_request`484- Verify that no wrapper script is swallowing stdout/stderr485 486## Repository Integration487 488Repository-level helper commands:489 490- `npm run test:sdk:python`491- `npm run lint:sdk:python`492- `npm run typecheck:sdk:python`493- `npm run smoke:sdk:python -- --qwen qwen`494 495## Real E2E Smoke496 497For a real runtime check (actual `qwen` process + real model call), run from498the repository root. The npm helper uses `python3`, so ensure it resolves to a499Python `>=3.10` interpreter:500 501```bash502npm run smoke:sdk:python -- --qwen qwen503```504 505This script runs:506 507- async single-turn query508- async control flow (`supported_commands`, permission mode updates)509- sync `query_sync` query510 511It prints JSON and returns non-zero on failure.512