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
6 changes: 5 additions & 1 deletion src/claude_agent_sdk/_internal/transport/subprocess_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -342,9 +342,13 @@ async def connect(self) -> None:

cmd = self._build_command()
try:
# Remove CLAUDECODE from parent environment to prevent nesting detection
# This allows SDK usage from within Claude Code (hooks, plugins, subagents)
parent_env = {k: v for k, v in os.environ.items() if k != "CLAUDECODE"}

# Merge environment variables: system -> user -> SDK required
process_env = {
**os.environ,
**parent_env,
**self._options.env, # User-provided env vars
"CLAUDE_CODE_ENTRYPOINT": "sdk-py",
"CLAUDE_AGENT_SDK_VERSION": __version__,
Expand Down
112 changes: 112 additions & 0 deletions tests/test_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -452,6 +452,118 @@ async def _test():

anyio.run(_test)

def test_claudecode_env_var_is_filtered(self):
"""Test that CLAUDECODE env var is filtered from subprocess to prevent nesting detection."""

async def _test():
# Simulate running inside Claude Code
os.environ["CLAUDECODE"] = "1"

options = make_options()

# Mock the subprocess to capture the env argument
with patch(
"anyio.open_process", new_callable=AsyncMock
) as mock_open_process:
# 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

# Return version process first, then main process
mock_open_process.side_effect = [mock_version_process, mock_process]

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

await transport.connect()

# Verify open_process was called twice (version check + main process)
assert mock_open_process.call_count == 2

# Check the second call (main process) for env vars
second_call_kwargs = mock_open_process.call_args_list[1].kwargs
assert "env" in second_call_kwargs
env_passed = second_call_kwargs["env"]

# CLAUDECODE should NOT be in env (filtered to prevent nesting detection)
assert "CLAUDECODE" not in env_passed

# But other vars should be present
assert "CLAUDE_CODE_ENTRYPOINT" in env_passed
assert env_passed["CLAUDE_CODE_ENTRYPOINT"] == "sdk-py"

anyio.run(_test)

def test_claudecode_can_be_explicitly_set(self):
"""Test that CLAUDECODE can be explicitly set via options env if needed."""

async def _test():
# Simulate running inside Claude Code
os.environ["CLAUDECODE"] = "1"

# User explicitly wants CLAUDECODE set (unusual but allowed)
options = make_options(env={"CLAUDECODE": "1"})

# Mock the subprocess to capture the env argument
with patch(
"anyio.open_process", new_callable=AsyncMock
) as mock_open_process:
# 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

# Return version process first, then main process
mock_open_process.side_effect = [mock_version_process, mock_process]

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

await transport.connect()

# Verify open_process was called twice
assert mock_open_process.call_count == 2

# Check the second call (main process) for env vars
second_call_kwargs = mock_open_process.call_args_list[1].kwargs
assert "env" in second_call_kwargs
env_passed = second_call_kwargs["env"]

# CLAUDECODE SHOULD be in env because user explicitly provided it
assert "CLAUDECODE" in env_passed
assert env_passed["CLAUDECODE"] == "1"

anyio.run(_test)

def test_connect_as_different_user(self):
"""Test connect as different user."""

Expand Down