mirrors/claude-code-sdk-python
1
0
Fork 0
mirror of https://github.com/anthropics/claude-code-sdk-python.git synced 2025-12-23 09:19:52 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Ruff

Browse source
This commit is contained in:
Dickson Tsai 2025-07-19 15:21:02 -07:00
parent 712948c2e7
commit c95c077b9b
No known key found for this signature in database
4 changed files with 18 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/claude_code_sdk/__init__.py
View file

@ -1,6 +1,5 @@
"""Claude SDK for Python."""
from ._errors import (
ClaudeSDKError,
CLIConnectionError,

4
src/claude_code_sdk/_internal/client.py
View file

@ -20,9 +20,7 @@ class InternalClient:
"""Process a query through transport."""
transport = SubprocessCLITransport(
prompt=prompt,
options=options,
close_stdin_after_prompt=True
prompt=prompt, options=options, close_stdin_after_prompt=True
)
try:

13
src/claude_code_sdk/_internal/transport/subprocess_cli.py
View file

@ -161,6 +161,7 @@ class SubprocessCLITransport(Transport):
self._stdin_stream = TextSendStream(self._process.stdin)
# Start streaming messages to stdin in background
import asyncio
asyncio.create_task(self._stream_to_stdin())
else:
# String mode: close stdin immediately (backward compatible)
@ -214,7 +215,7 @@ class SubprocessCLITransport(Transport):
"type": "user",
"message": {"role": "user", "content": str(message)},
"parent_tool_use_id": None,
"session_id": options.get("session_id", "default")
"session_id": options.get("session_id", "default"),
}
await self._stdin_stream.send(json.dumps(message) + "\n")
@ -362,7 +363,9 @@ class SubprocessCLITransport(Transport):
async def interrupt(self) -> None:
"""Send interrupt control request (only works in streaming mode)."""
if not self._is_streaming:
raise CLIConnectionError("Interrupt requires streaming mode (AsyncIterable prompt)")
raise CLIConnectionError(
"Interrupt requires streaming mode (AsyncIterable prompt)"
)
if not self._stdin_stream:
raise CLIConnectionError("Not connected or stdin not available")
@ -382,7 +385,7 @@ class SubprocessCLITransport(Transport):
control_request = {
"type": "control_request",
"request_id": request_id,
"request": request
"request": request,
}
# Send request
@ -397,7 +400,9 @@ class SubprocessCLITransport(Transport):
response = self._pending_control_responses.pop(request_id)
if response.get("subtype") == "error":
raise CLIConnectionError(f"Control request failed: {response.get('error')}")
raise CLIConnectionError(
f"Control request failed: {response.get('error')}"
)
return response
except TimeoutError:

12
tests/test_streaming_client.py
View file

@ -389,7 +389,8 @@ class TestQueryWithAsyncIterable:
if response is None:
response = '{"type": "result", "subtype": "success", "duration_ms": 100, "duration_api_ms": 50, "is_error": false, "num_turns": 1, "session_id": "test", "total_cost_usd": 0.001}'
script_content = textwrap.dedent("""
script_content = textwrap.dedent(
"""
#!/usr/bin/env python3
import sys
import json
@ -416,7 +417,8 @@ class TestQueryWithAsyncIterable:
)
if expected_messages is not None:
script_content += textwrap.dedent(f"""
script_content += textwrap.dedent(
f"""
# Verify we got the expected messages
assert len(stdin_messages) == {len(expected_messages)}
""",
@ -425,12 +427,14 @@ class TestQueryWithAsyncIterable:
script_content += f'''assert '"{msg}"' in stdin_messages[{i}]\n'''
if should_error:
script_content += textwrap.dedent("""
script_content += textwrap.dedent(
"""
sys.exit(1)
""",
)
else:
script_content += textwrap.dedent(f"""
script_content += textwrap.dedent(
f"""
# Output response
print('{response}')
""",