diff --git a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py index 4b47f115..abf3ffaa 100644 --- a/src/claude_agent_sdk/_internal/transport/subprocess_cli.py +++ b/src/claude_agent_sdk/_internal/transport/subprocess_cli.py @@ -663,3 +663,31 @@ async def _check_claude_version(self) -> None: def is_ready(self) -> bool: """Check if transport is ready for communication.""" return self._ready + + @property + def pid(self) -> int | None: + """OS process ID of the spawned Claude Code CLI subprocess, or None if not running. + + Exposed for callers that need to implement external cleanup (e.g. + ``os.killpg``, ``psutil.Process``, or cgroup management) beyond what + :meth:`close` provides. Returns None before :meth:`connect` has been + called and after :meth:`close` has completed. + """ + if self._process is None: + return None + return self._process.pid + + @property + def process(self) -> Process | None: + """The underlying ``anyio.abc.Process`` handle, or None if not running. + + Exposed for advanced cleanup or signal handling. Prefer :meth:`close` + for normal shutdown — it already implements bounded-wait graceful + termination followed by SIGKILL fallback. Use this property only when + you need finer control (e.g. sending SIGTERM to an entire process + group, or attaching observability tooling to the subprocess). + + Returns None before :meth:`connect` has been called and after + :meth:`close` has completed. + """ + return self._process diff --git a/tests/test_transport.py b/tests/test_transport.py index b2c40923..b7d3762a 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -62,6 +62,31 @@ def test_init_uses_provided_cli_path(self): ) assert transport._cli_path == "/usr/bin/claude" + def test_pid_returns_none_before_connect(self): + """pid is None before connect() spawns the subprocess.""" + transport = SubprocessCLITransport(prompt="test", options=make_options()) + assert transport.pid is None + + def test_process_returns_none_before_connect(self): + """process is None before connect() spawns the subprocess.""" + transport = SubprocessCLITransport(prompt="test", options=make_options()) + assert transport.process is None + + def test_pid_returns_subprocess_pid_when_running(self): + """pid returns the underlying subprocess PID when the process exists.""" + transport = SubprocessCLITransport(prompt="test", options=make_options()) + mock_process = MagicMock() + mock_process.pid = 12345 + transport._process = mock_process + assert transport.pid == 12345 + + def test_process_returns_process_when_running(self): + """process returns the underlying anyio Process handle when it exists.""" + transport = SubprocessCLITransport(prompt="test", options=make_options()) + mock_process = MagicMock() + transport._process = mock_process + assert transport.process is mock_process + def test_build_command_basic(self): """Test building basic CLI command.""" transport = SubprocessCLITransport(prompt="Hello", options=make_options())