Skip to content
Open
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
21 changes: 12 additions & 9 deletions src/claude_agent_sdk/_internal/transport/subprocess_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -366,15 +366,18 @@ async def connect(self) -> None:
# For backward compat: use debug_stderr file object if no callback and debug is on
stderr_dest = PIPE if should_pipe_stderr else None

self._process = await anyio.open_process(
cmd,
stdin=PIPE,
stdout=PIPE,
stderr=stderr_dest,
cwd=self._cwd,
env=process_env,
user=self._options.user,
)
# user parameter is not supported on Windows
process_kwargs: dict[str, object] = {
"stdin": PIPE,
"stdout": PIPE,
"stderr": stderr_dest,
"cwd": self._cwd,
"env": process_env,
}
if platform.system() != "Windows" and self._options.user is not None:
process_kwargs["user"] = self._options.user

self._process = await anyio.open_process(cmd, **process_kwargs)

if self._process.stdout:
self._stdout_stream = TextReceiveStream(self._process.stdout)
Expand Down
48 changes: 48 additions & 0 deletions tests/test_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -503,6 +503,54 @@ async def _test():

anyio.run(_test)

def test_connect_user_param_skipped_on_windows(self):
"""Test that user parameter is not passed to open_process on Windows."""

async def _test():
options = make_options(user="claude")

with (
patch(
"anyio.open_process", new_callable=AsyncMock
) as mock_open_process,
patch(
"claude_agent_sdk._internal.transport.subprocess_cli.platform"
) as mock_platform,
):
mock_platform.system.return_value = "Windows"

# Mock version check process
mock_version_process = MagicMock()
mock_version_process.stdout = MagicMock()
mock_version_process.stdout.receive = AsyncMock(
return_value=b"2.0.0 (Claude Code)"
)
mock_version_process.terminate = MagicMock()
mock_version_process.wait = AsyncMock()

# Mock main process
mock_process = MagicMock()
mock_process.stdout = MagicMock()
mock_stdin = MagicMock()
mock_stdin.aclose = AsyncMock()
mock_process.stdin = mock_stdin
mock_process.returncode = None

mock_open_process.side_effect = [mock_version_process, mock_process]

transport = SubprocessCLITransport(
prompt="test",
options=options,
)

await transport.connect()

# Check the second call (main process) does NOT include user
second_call_kwargs = mock_open_process.call_args_list[1].kwargs
assert "user" not in second_call_kwargs

anyio.run(_test)

def test_build_command_with_sandbox_only(self):
"""Test building CLI command with sandbox settings (no existing settings)."""
import json
Expand Down