feat: add stderr callback to capture CLI debug output (#170)

## Summary
- Add stderr callback option to ClaudeCodeOptions to capture CLI
subprocess stderr output
- Matches TypeScript SDK's stderr callback behavior for feature parity
- Useful for debugging and monitoring CLI operations

## Changes
- Added `stderr: Callable[[str], None] | None` field to
`ClaudeCodeOptions`
- Updated `SubprocessCLITransport` to handle stderr streaming with async
task
- Added example demonstrating stderr callback usage
- Added e2e tests to verify functionality

## Test plan
- [x] Run e2e tests: `python -m pytest e2e-tests/test_stderr_callback.py
-v`
- [x] Run example: `python examples/stderr_callback_example.py`
- [x] Verify backward compatibility with existing `debug_stderr` field
- [x] All linting and type checks pass

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Claude <[email protected]>

Author: dicksontsai

e2e-tests/test_stderr_callback.py

@@ -0,0 +1,49 @@
+"""End-to-end test for stderr callback functionality."""
+
+import pytest
+
+from claude_code_sdk import ClaudeCodeOptions, query
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_stderr_callback_captures_debug_output():
+    """Test that stderr callback receives debug output when enabled."""
+    stderr_lines = []
+
+    def capture_stderr(line: str):
+        stderr_lines.append(line)
+
+    # Enable debug mode to generate stderr output
+    options = ClaudeCodeOptions(
+        stderr=capture_stderr,
+        extra_args={"debug-to-stderr": None}
+    )
+
+    # Run a simple query
+    async for _ in query(prompt="What is 1+1?", options=options):
+        pass  # Just consume messages
+
+    # Verify we captured debug output
+    assert len(stderr_lines) > 0, "Should capture stderr output with debug enabled"
+    assert any("[DEBUG]" in line for line in stderr_lines), "Should contain DEBUG messages"
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_stderr_callback_without_debug():
+    """Test that stderr callback works but receives no output without debug mode."""
+    stderr_lines = []
+
+    def capture_stderr(line: str):
+        stderr_lines.append(line)
+
+    # No debug mode enabled
+    options = ClaudeCodeOptions(stderr=capture_stderr)
+
+    # Run a simple query
+    async for _ in query(prompt="What is 1+1?", options=options):
+        pass  # Just consume messages
+
+    # Should work but capture minimal/no output without debug
+    assert len(stderr_lines) == 0, "Should not capture stderr output without debug mode"

examples/stderr_callback_example.py

@@ -0,0 +1,44 @@
+"""Simple example demonstrating stderr callback for capturing CLI debug output."""
+
+import asyncio
+
+from claude_code_sdk import ClaudeCodeOptions, query
+
+
+async def main():
+    """Capture stderr output from the CLI using a callback."""
+
+    # Collect stderr messages
+    stderr_messages = []
+
+    def stderr_callback(message: str):
+        """Callback that receives each line of stderr output."""
+        stderr_messages.append(message)
+        # Optionally print specific messages
+        if "[ERROR]" in message:
+            print(f"Error detected: {message}")
+
+    # Create options with stderr callback and enable debug mode
+    options = ClaudeCodeOptions(
+        stderr=stderr_callback,
+        extra_args={"debug-to-stderr": None}  # Enable debug output
+    )
+
+    # Run a query
+    print("Running query with stderr capture...")
+    async for message in query(
+        prompt="What is 2+2?",
+        options=options
+    ):
+        if hasattr(message, 'content'):
+            if isinstance(message.content, str):
+                print(f"Response: {message.content}")
+
+    # Show what we captured
+    print(f"\nCaptured {len(stderr_messages)} stderr lines")
+    if stderr_messages:
+        print("First stderr line:", stderr_messages[0][:100])
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/claude_code_sdk/_internal/transport/subprocess_cli.py

@@ -12,6 +12,7 @@
 from typing import Any
 
 import anyio
+import anyio.abc
 from anyio.abc import Process
 from anyio.streams.text import TextReceiveStream, TextSendStream
 
@@ -43,6 +44,8 @@ def __init__(
         self._process: Process | None = None
         self._stdout_stream: TextReceiveStream | None = None
         self._stdin_stream: TextSendStream | None = None
+        self._stderr_stream: TextReceiveStream | None = None
+        self._stderr_task_group: anyio.abc.TaskGroup | None = None
         self._ready = False
         self._exit_error: Exception | None = None  # Track process exit errors
 
@@ -216,14 +219,15 @@ async def connect(self) -> None:
             if self._cwd:
                 process_env["PWD"] = self._cwd
 
-            # Only output stderr if customer explicitly requested debug output and provided a file object
-            stderr_dest = (
-                self._options.debug_stderr
-                if "debug-to-stderr" in self._options.extra_args
-                and self._options.debug_stderr
-                else None
+            # Pipe stderr if we have a callback OR debug mode is enabled
+            should_pipe_stderr = (
+                self._options.stderr is not None
+                or "debug-to-stderr" in self._options.extra_args
             )
 
+            # 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,
@@ -237,6 +241,14 @@ async def connect(self) -> None:
             if self._process.stdout:
                 self._stdout_stream = TextReceiveStream(self._process.stdout)
 
+            # Setup stderr stream if piped
+            if should_pipe_stderr and self._process.stderr:
+                self._stderr_stream = TextReceiveStream(self._process.stderr)
+                # Start async task to read stderr
+                self._stderr_task_group = anyio.create_task_group()
+                await self._stderr_task_group.__aenter__()
+                self._stderr_task_group.start_soon(self._handle_stderr)
+
             # Setup stdin for streaming mode
             if self._is_streaming and self._process.stdin:
                 self._stdin_stream = TextSendStream(self._process.stdin)
@@ -262,19 +274,59 @@ async def connect(self) -> None:
             self._exit_error = error
             raise error from e
 
+    async def _handle_stderr(self) -> None:
+        """Handle stderr stream - read and invoke callbacks."""
+        if not self._stderr_stream:
+            return
+
+        try:
+            async for line in self._stderr_stream:
+                line_str = line.rstrip()
+                if not line_str:
+                    continue
+
+                # Call the stderr callback if provided
+                if self._options.stderr:
+                    self._options.stderr(line_str)
+
+                # For backward compatibility: write to debug_stderr if in debug mode
+                elif (
+                    "debug-to-stderr" in self._options.extra_args
+                    and self._options.debug_stderr
+                ):
+                    self._options.debug_stderr.write(line_str + "\n")
+                    if hasattr(self._options.debug_stderr, "flush"):
+                        self._options.debug_stderr.flush()
+        except anyio.ClosedResourceError:
+            pass  # Stream closed, exit normally
+        except Exception:
+            pass  # Ignore other errors during stderr reading
+
     async def close(self) -> None:
         """Close the transport and clean up resources."""
         self._ready = False
 
         if not self._process:
             return
 
+        # Close stderr task group if active
+        if self._stderr_task_group:
+            with suppress(Exception):
+                self._stderr_task_group.cancel_scope.cancel()
+                await self._stderr_task_group.__aexit__(None, None, None)
+            self._stderr_task_group = None
+
         # Close streams
         if self._stdin_stream:
             with suppress(Exception):
                 await self._stdin_stream.aclose()
             self._stdin_stream = None
 
+        if self._stderr_stream:
+            with suppress(Exception):
+                await self._stderr_stream.aclose()
+            self._stderr_stream = None
+
         if self._process.stdin:
             with suppress(Exception):
                 await self._process.stdin.aclose()
@@ -291,6 +343,7 @@ async def close(self) -> None:
         self._process = None
         self._stdout_stream = None
         self._stdin_stream = None
+        self._stderr_stream = None
         self._exit_error = None
 
     async def write(self, data: str) -> None:

src/claude_code_sdk/types.py

@@ -322,7 +322,8 @@ class ClaudeAgentOptions:
     )  # Pass arbitrary CLI flags
     debug_stderr: Any = (
         sys.stderr
-    )  # File-like object for debug output when debug-to-stderr is set
+    )  # Deprecated: File-like object for debug output. Use stderr callback instead.
+    stderr: Callable[[str], None] | None = None  # Callback for stderr output from CLI
 
     # Tool permission callback
     can_use_tool: CanUseTool | None = None