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
72 changes: 66 additions & 6 deletions src/claude_code_sdk/_internal/transport/subprocess_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ def __init__(
self._process: Process | None = None
self._stdout_stream: TextReceiveStream | None = None
self._stderr_stream: TextReceiveStream | None = None
self._received_any_output = False

def _find_cli(self) -> str:
"""Find Claude Code CLI binary."""
Expand Down Expand Up @@ -75,6 +76,10 @@ def _find_cli(self) -> str:
def _build_command(self) -> list[str]:
"""Build CLI command with arguments."""
cmd = [self._cli_path, "--output-format", "stream-json", "--verbose"]

# Add debug flag when using bypassPermissions to capture permission warnings
if self._options.permission_mode == "bypassPermissions":
cmd.append("--debug")

if self._options.system_prompt:
cmd.extend(["--system-prompt", self._options.system_prompt])
Expand Down Expand Up @@ -129,7 +134,7 @@ async def connect(self) -> None:
stdout=PIPE,
stderr=PIPE,
cwd=self._cwd,
env={**os.environ, "CLAUDE_CODE_ENTRYPOINT": "sdk-py"},
env={**os.environ, "CLAUDE_CODE_ENTRYPOINT": "sdk-cli"},
)

if self._process.stdout:
Expand Down Expand Up @@ -214,6 +219,7 @@ async def read_stderr() -> None:
try:
data = json.loads(json_buffer)
json_buffer = ""
self._received_any_output = True
try:
yield data
except GeneratorExit:
Expand All @@ -225,14 +231,68 @@ async def read_stderr() -> None:
pass

await self._process.wait()
if self._process.returncode is not None and self._process.returncode != 0:
stderr_output = "\n".join(stderr_lines)
if stderr_output and "error" in stderr_output.lower():
stderr_output = "\n".join(stderr_lines)

# First check for specific error messages in stderr
if self._options.permission_mode == "bypassPermissions":
# Check for known bypassPermissions issues
if "cannot be used with root/sudo privileges" in stderr_output:
raise ProcessError(
"CLI process failed",
exit_code=self._process.returncode,
"bypassPermissions mode cannot be used when running as root/sudo. "
"This is a security restriction in Claude Code. To fix this:\n"
"1. Run as a non-root user, or\n"
"2. Use permission_mode='acceptEdits' instead",
exit_code=self._process.returncode or 0,
stderr=stderr_output,
)

if "bypassPermissions mode is disabled" in stderr_output:
raise ProcessError(
"bypassPermissions mode is disabled by Claude Code settings. "
"The CLI will fall back to default permissions which requires user input, "
"but the SDK runs without stdin. To fix this, either:\n"
"1. Enable bypassPermissions in your Claude Code settings, or\n"
"2. Use permission_mode='acceptEdits' instead",
exit_code=self._process.returncode or 0,
stderr=stderr_output,
)

# Then check for the general "no output" case
if not self._received_any_output:
# This can happen for various reasons depending on permission mode
permission_mode = self._options.permission_mode or "default"

if permission_mode == "bypassPermissions":
# Special handling for bypassPermissions silent failures
raise ProcessError(
"Claude Code CLI terminated without producing any output when using "
"bypassPermissions mode. This can happen when:\n"
"1. Running as root/sudo (security restriction)\n"
"2. bypassPermissions is disabled in settings\n"
"3. Other security restrictions are in place\n\n"
"Try using permission_mode='acceptEdits' instead, which is "
"the recommended mode for SDK usage.",
exit_code=self._process.returncode or 0,
stderr=stderr_output if stderr_output else None,
)
elif permission_mode == "default":
raise ProcessError(
"Claude Code CLI appears to have terminated without output. "
"This often happens when the CLI is waiting for interactive "
"permission prompts that cannot be answered in SDK mode. "
"Try using permission_mode='acceptEdits' or 'bypassPermissions'.",
exit_code=self._process.returncode or 0,
stderr=stderr_output if stderr_output else None,
)

# Finally, handle any other non-zero exit codes
if self._process.returncode is not None and self._process.returncode != 0:
# Always raise an error for non-zero exit codes
raise ProcessError(
f"CLI process failed with exit code {self._process.returncode}",
exit_code=self._process.returncode,
stderr=stderr_output if stderr_output else None,
)

def is_connected(self) -> bool:
"""Check if subprocess is running."""
Expand Down
176 changes: 176 additions & 0 deletions tests/test_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,3 +132,179 @@ def test_receive_messages(self):
# So we just verify the transport can be created and basic structure is correct
assert transport._prompt == "test"
assert transport._cli_path == "/usr/bin/claude"

def test_bypass_permissions_disabled_error(self):
"""Test that bypassPermissions being disabled raises proper error."""
from claude_code_sdk._errors import ProcessError
from anyio.streams.text import TextReceiveStream
from anyio import ClosedResourceError

async def _test():
with patch("anyio.open_process") as mock_exec:
# Create a mock process that simulates the CLI behavior
mock_process = MagicMock()
mock_process.returncode = 0 # CLI exits successfully
mock_process.wait = AsyncMock()

# Create mock stdout/stderr streams
mock_stdout_stream = AsyncMock()
mock_stderr_stream = AsyncMock()

# Simulate empty stdout (no JSON messages) followed by closed stream
async def stdout_iter(self):
raise ClosedResourceError()
yield # This won't be reached

# Simulate stderr with the warning message
async def stderr_iter(self):
yield "[ERROR] bypassPermissions mode is disabled by settings"
raise ClosedResourceError()

type(mock_stdout_stream).__aiter__ = stdout_iter
type(mock_stderr_stream).__aiter__ = stderr_iter

mock_process.stdout = MagicMock()
mock_process.stderr = MagicMock()
mock_exec.return_value = mock_process

# Patch TextReceiveStream to return our mocks
with patch("claude_code_sdk._internal.transport.subprocess_cli.TextReceiveStream") as mock_text_stream:
mock_text_stream.side_effect = [mock_stdout_stream, mock_stderr_stream]

transport = SubprocessCLITransport(
prompt="test",
options=ClaudeCodeOptions(permission_mode="bypassPermissions"),
cli_path="/usr/bin/claude",
)

await transport.connect()

with pytest.raises(ProcessError) as exc_info:
# Consume all messages to trigger the error check
async for _ in transport.receive_messages():
pass

assert "bypassPermissions mode is disabled" in str(exc_info.value)
assert "requires user input" in str(exc_info.value)
assert "acceptEdits" in str(exc_info.value)

anyio.run(_test)

def test_bypass_permissions_adds_debug_flag(self):
"""Test that bypassPermissions mode adds --debug flag to command."""
transport = SubprocessCLITransport(
prompt="test",
options=ClaudeCodeOptions(permission_mode="bypassPermissions"),
cli_path="/usr/bin/claude",
)

cmd = transport._build_command()
assert "--debug" in cmd

def test_bypass_permissions_root_user_error(self):
"""Test that running as root with bypassPermissions raises proper error."""
from claude_code_sdk._errors import ProcessError
from anyio import ClosedResourceError

async def _test():
with patch("anyio.open_process") as mock_exec:
# Create a mock process that exits with code 1 (root user rejection)
mock_process = MagicMock()
mock_process.returncode = 1
mock_process.wait = AsyncMock()

# Create mock stdout/stderr streams
mock_stdout_stream = AsyncMock()
mock_stderr_stream = AsyncMock()

# Simulate empty stdout
async def stdout_iter(self):
raise ClosedResourceError()
yield # Won't be reached

# Simulate stderr with root user error
async def stderr_iter(self):
yield "--dangerously-skip-permissions cannot be used with root/sudo privileges for security reasons"
raise ClosedResourceError()

type(mock_stdout_stream).__aiter__ = stdout_iter
type(mock_stderr_stream).__aiter__ = stderr_iter

mock_process.stdout = MagicMock()
mock_process.stderr = MagicMock()
mock_exec.return_value = mock_process

# Patch TextReceiveStream to return our mocks
with patch("claude_code_sdk._internal.transport.subprocess_cli.TextReceiveStream") as mock_text_stream:
mock_text_stream.side_effect = [mock_stdout_stream, mock_stderr_stream]

transport = SubprocessCLITransport(
prompt="test",
options=ClaudeCodeOptions(permission_mode="bypassPermissions"),
cli_path="/usr/bin/claude",
)

await transport.connect()

with pytest.raises(ProcessError) as exc_info:
async for _ in transport.receive_messages():
pass

assert "cannot be used when running as root/sudo" in str(exc_info.value)
assert "Run as a non-root user" in str(exc_info.value)

anyio.run(_test)

def test_bypass_permissions_no_output(self):
"""Test bypassPermissions mode that exits without any output."""
from claude_code_sdk._errors import ProcessError
from anyio import ClosedResourceError

async def _test():
with patch("anyio.open_process") as mock_exec:
# Create a mock process that exits cleanly but produces no output
mock_process = MagicMock()
mock_process.returncode = 0 # Clean exit
mock_process.wait = AsyncMock()

# Create mock stdout/stderr streams that immediately close
mock_stdout_stream = AsyncMock()
mock_stderr_stream = AsyncMock()

# Both streams immediately close without yielding anything
async def empty_iter(self):
raise ClosedResourceError()
yield # Won't be reached

type(mock_stdout_stream).__aiter__ = empty_iter
type(mock_stderr_stream).__aiter__ = empty_iter

mock_process.stdout = MagicMock()
mock_process.stderr = MagicMock()
mock_exec.return_value = mock_process

# Patch TextReceiveStream to return our mocks
with patch("claude_code_sdk._internal.transport.subprocess_cli.TextReceiveStream") as mock_text_stream:
mock_text_stream.side_effect = [mock_stdout_stream, mock_stderr_stream]

transport = SubprocessCLITransport(
prompt="test",
options=ClaudeCodeOptions(permission_mode="bypassPermissions"),
cli_path="/usr/bin/claude",
)

await transport.connect()

with pytest.raises(ProcessError) as exc_info:
# This simulates the "no response" issue - the iterator completes
# without yielding any messages
async for _ in transport.receive_messages():
pass

error_msg = str(exc_info.value)
assert "terminated without producing any output" in error_msg
assert "bypassPermissions mode" in error_msg
assert "Running as root/sudo" in error_msg
assert "acceptEdits" in error_msg

anyio.run(_test)
Loading