Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions packages/sdk-python/src/qwen_code_sdk/protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -238,6 +238,11 @@ class CLIControlSupportedCommandsRequest(TypedDict):
subtype: Literal["supported_commands"]


class CLIControlGetContextUsageRequest(TypedDict):
subtype: Literal["get_context_usage"]
show_details: NotRequired[bool]


ControlRequestPayload: TypeAlias = (
CLIControlInterruptRequest
| CLIControlPermissionRequest
Expand All @@ -246,6 +251,7 @@ class CLIControlSupportedCommandsRequest(TypedDict):
| CLIControlSetModelRequest
| CLIControlMcpStatusRequest
| CLIControlSupportedCommandsRequest
| CLIControlGetContextUsageRequest
| dict[str, Any]
)

Expand Down
8 changes: 8 additions & 0 deletions packages/sdk-python/src/qwen_code_sdk/query.py
Original file line number Diff line number Diff line change
Expand Up @@ -482,6 +482,14 @@ async def mcp_server_status(self) -> dict[str, Any] | None:
await self._ensure_started()
return await self._send_control_request("mcp_server_status")

async def get_context_usage(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] The async Query class now has get_context_usage(), but SyncQuery in sync_query.py is missing the corresponding sync wrapper. Every other control-request method (supported_commands, mcp_server_status, set_model, set_permission_mode) has a sync counterpart. Sync SDK consumers currently have no way to call this new feature.

Consider adding to SyncQuery (after mcp_server_status):

def get_context_usage(self, show_details: bool = False) -> Any:
    q = self._require_query()
    return asyncio.run_coroutine_threadsafe(
        q.get_context_usage(show_details),
        self._loop,
    ).result(timeout=q.control_request_timeout + _SYNC_TIMEOUT_MARGIN)

— qwen3.7-max via Qwen Code /review

self, show_details: bool = False
) -> dict[str, Any] | None:
await self._ensure_started()
return await self._send_control_request(
"get_context_usage", {"show_details": show_details}
)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] No tests were added for get_context_usage(). The TS SDK has dedicated unit tests (Query.test.ts lines 1211-1287) covering the request payload, show_details parameter forwarding, response deserialization, and closed-query error handling. The Python SDK has analogous tests for supported_commands() and mcp_server_status() in test_query_core.py and integration tests.

Additionally, the integration test mock in tests/integration/conftest.py has no handler for the get_context_usage subtype — it falls through to the error branch, so integration tests cannot exercise this path.

Suggested additions:

  • Unit test in test_query_core.py verifying the outgoing payload contains {"subtype": "get_context_usage", "show_details": True/False} and that a mock response is returned correctly
  • A get_context_usage branch in conftest.py's mock CLI handler returning a representative response
  • Optionally, an integration test in test_async_query.py

— qwen3.7-max via Qwen Code /review


@property
def control_request_timeout(self) -> float:
return self._options.timeout.control_request
Expand Down