* Codex worktree snapshot: startup-cleanup Co-authored-by: Codex * Add Python SDK real smoke test Adds a repository-only real E2E smoke script for the Python SDK, plus npm and developer documentation entry points. Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): address review findings — bugs, type safety, and test coverage - Fix prepare_spawn_info: JS files now use "node" instead of sys.executable - Fix protocol.py: correct total=False misuse on 7 TypedDicts (required fields were optional) - Fix query.py: add _closed guard in _ensure_started, suppress exceptions in close() - Fix sync_query.py: prevent close() deadlock, add context manager, add timeouts - Fix transport.py: handle malformed JSON lines, add _closed guard in start() - Fix validation.py: use uuid.RFC_4122 instead of magic UUID - Fix __init__.py: export TextBlock, widen query_sync signature - Remove dead code: ensure_not_aborted, write_json_line, _thread_error - Add 12 new tests (29 → 41): context managers, JSON skip, closed guards, spawn info, timeouts Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): address wenshao review — session_id, bool validation, debug stderr - Fix continue_session=True generating a wrong random session_id - Add _as_optional_bool helper for strict type validation on bool fields - Default debug stderr to sys.stderr when no custom callback is provided Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): address remaining wenshao review feedback Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * test(cli): harden settings dialog restart prompt test Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): review fixes — UUID compat, stderr fallback, sync cleanup - Remove UUID version restriction to support v6/v7/v8 (RFC 9562) - Always write to sys.stderr when stderr callback raises (was silent when debug=False) - Prevent duplicate _STOP sentinel in SyncQuery.close() via _stop_sent flag - Add ruff format --check to CI workflow - Fix smoke_real.py version guard: fail early before imports instead of NameError - Apply ruff format to existing files Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): remaining review fixes — exit_code attr, guard strictness, sync timeout - Add exit_code attribute to ProcessExitError for programmatic access - Strengthen is_control_response/is_control_cancel guards to require payload fields, preventing misrouting of malformed messages - Expose control_request_timeout property on Query so SyncQuery uses the configured timeout instead of a hardcoded 30s default - Use dataclasses.replace() instead of direct mutation on frozen-style QueryOptions in query() factory - Add ResourceWarning in SyncQuery.__del__ when not properly closed Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): add exit_code default and guard __del__ against partial GC - Give ProcessExitError.exit_code a default value (-1) so user code can construct the exception with just a message string - Wrap SyncQuery.__del__ in try/except AttributeError to prevent crashes when the object is partially garbage-collected Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): review fixes — resource leak, type safety, CI matrix, docs - Fix SyncQuery.__del__ to call close() on GC instead of only warning - Replace hasattr duck-type check with isinstance(prompt, AsyncIterable) - Type-validate permission_mode/auth_type in QueryOptions.from_mapping - Use TypeGuard return types on all is_sdk_*/is_control_* predicates - Add 5s margin to sync wrapper timeouts to prevent error type masking - Expand CI matrix to test Python 3.10, 3.11, 3.12 - Change ProcessExitError.exit_code default from -1 to None - Add stderr to docs QueryOptions listing - Update README sync example to use context manager pattern Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): preserve iterator exhaustion state and suppress detached task warning - Add _exhausted flag to Query.__anext__ and SyncQuery.__next__ so repeated iteration after end-of-stream raises Stop(Async)Iteration instead of blocking forever. - Remove re-raise in _initialize() to prevent asyncio "Task exception was never retrieved" warning on detached tasks; the error is already surfaced via _finish_with_error(). Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): reject mcp_servers at validation time and add iterator/init tests - Reject mcp_servers in validate_query_options() with a clear error instead of advertising MCP support to the CLI and then failing at runtime when mcp_message arrives. - Remove dead mcp_servers branch from _initialize(). - Add tests for async/sync iterator exhaustion, detached init task warning suppression, and mcp_servers validation. Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk-python): fix ruff lint errors in new tests - Use ControlRequestTimeoutError instead of bare Exception (B017) - Fix import sorting for stdlib vs third-party (I001) - Break long line to stay within 88-char limit (E501) Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * style(sdk-python): apply ruff format to new tests Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> --------- Co-authored-by: jinye.djy <jinye.djy@alibaba-inc.com> Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
3.8 KiB
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
qwenexecutable is required in v1 - Transport scope: process transport only
- Not included in v1: ACP transport, SDK-embedded MCP servers
Install
pip install qwen-code-sdk
If qwen is not on PATH, pass path_to_qwen_executable explicitly.
Quick Start
import asyncio
from qwen_code_sdk import is_sdk_result_message, query
async def main() -> None:
result = 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"])
asyncio.run(main())
API Surface
Top-level entry points
query(prompt, options=None) -> Queryquery_sync(prompt, options=None) -> SyncQuery
prompt supports either:
strfor single-turn requestsAsyncIterable[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
Supported options in v1:
cwdmodelpath_to_qwen_executablepermission_modecan_use_toolenvsystem_promptappend_system_promptdebugmax_session_turnscore_toolsexclude_toolsallowed_toolsauth_typeinclude_partial_messagesresumecontinue_sessionsession_idtimeoutmcp_serversstderr
Session argument priority is fixed as:
resumecontinue_sessionsession_id
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
- Timeout fallback: deny
- Callback exceptions: converted to deny with an error message
- Callback context:
cancel_event,suggestions, andblocked_path - Callback contract:
can_use_toolmust be async with 3 positional arguments;stderrmust accept 1 positional string argument
Error Model
ValidationError: invalid options, invalid UUIDs, unsupported combinationsControlRequestTimeoutError: initialize, interrupt, or other control request timed outProcessExitError: CLI exited non-zeroAbortError: control request or session was cancelled
Troubleshooting
If the SDK cannot start the CLI:
- Verify
qwen --versionworks in the target environment - Pass
path_to_qwen_executableif your shell usesnvm,pyenv, or other non-standard PATH setup - Use
debug=Trueorstderr=printto surface CLI stderr while debugging
If session control calls time out:
- Check that the target
qwenversion 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:pythonnpm run lint:sdk:pythonnpm run typecheck:sdk:pythonnpm 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:
npm run smoke:sdk:python -- --qwen qwen
This script runs:
- async single-turn query
- async control flow (
supported_commands, permission mode updates) - sync
query_syncquery
It prints JSON and returns non-zero on failure.