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
28 changes: 28 additions & 0 deletions src/claude_agent_sdk/_internal/transport/subprocess_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
25 changes: 25 additions & 0 deletions tests/test_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down