From 96e0a59b40d1820a15c242d9a039d749fa75a136 Mon Sep 17 00:00:00 2001 From: Mukunda Katta Date: Tue, 14 Apr 2026 23:36:57 -0700 Subject: [PATCH] feat(transport): expose pid and process as public properties MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses anthropics/anthropic-sdk-python#1370 (filed on the wrong repo — code lives here). SubprocessCLITransport spawns a subprocess but exposes no public way to inspect the process or its PID. Callers needing external cleanup (killpg, psutil, cgroups) have been walking async-generator frame internals to reach the underlying Process handle — a workaround that breaks on any internal refactor. Adds two read-only properties: - transport.pid -> OS process ID, or None before connect()/after close() - transport.process -> anyio.abc.Process handle, or None Both return None when no subprocess is running, so callers can safely check state without try/except. Note on other asks in #1370: - Bounded wait + SIGKILL fallback in close() is already implemented (see lines 481-497 of subprocess_cli.py). - start_new_session=True would change cleanup semantics (Windows has no POSIX sessions) and is out of scope for this minimal change. Tests (4 new): pid/process None before connect, pid/process return expected values when _process is set. Signed-off-by: Mukunda Katta --- .../_internal/transport/subprocess_cli.py | 28 +++++++++++++++++++ tests/test_transport.py | 25 +++++++++++++++++ 2 files changed, 53 insertions(+) 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())