Preview

src/claude_code_sdk/__init__.py

@@ -7,6 +7,7 @@
     CLINotFoundError,
     ProcessError,
 )
+from ._internal.transport import Transport
 from .client import ClaudeSDKClient
 from .query import query
 from .types import (
@@ -30,6 +31,8 @@
 __all__ = [
     # Main exports
     "query",
+    # Transport
+    "Transport",
     "ClaudeSDKClient",
     # Types
     "PermissionMode",

src/claude_code_sdk/_internal/client.py

@@ -3,8 +3,12 @@
 from collections.abc import AsyncIterable, AsyncIterator
 from typing import Any
 
-from ..types import ClaudeCodeOptions, Message
+from ..types import (
+    ClaudeCodeOptions,
+    Message,
+)
 from .message_parser import parse_message
+from .transport import Transport
 from .transport.subprocess_cli import SubprocessCLITransport
 
 
@@ -15,19 +19,26 @@ def __init__(self) -> None:
         """Initialize the internal client."""
 
     async def process_query(
-        self, prompt: str | AsyncIterable[dict[str, Any]], options: ClaudeCodeOptions
+        self,
+        prompt: str | AsyncIterable[dict[str, Any]],
+        options: ClaudeCodeOptions,
+        transport: Transport | None = None,
     ) -> AsyncIterator[Message]:
         """Process a query through transport."""
 
-        transport = SubprocessCLITransport(
-            prompt=prompt, options=options, close_stdin_after_prompt=True
-        )
+        # Use provided transport or choose one based on configuration
+        if transport is not None:
+            chosen_transport = transport
+        else:
+            chosen_transport = SubprocessCLITransport(
+                prompt=prompt, options=options, close_stdin_after_prompt=True
+            )
 
         try:
-            await transport.connect()
+            await chosen_transport.connect()
 
-            async for data in transport.receive_messages():
+            async for data in chosen_transport.receive_messages():
                 yield parse_message(data)
 
         finally:
-            await transport.disconnect()
+            await chosen_transport.disconnect()

src/claude_code_sdk/_internal/transport/__init__.py

@@ -6,7 +6,13 @@
 
 
 class Transport(ABC):
-    """Abstract transport for Claude communication."""
+    """Abstract transport for Claude communication.
+
+    WARNING: This internal API is exposed for custom transport implementations
+    (e.g., remote Claude Code connections). The Claude Code team may change or
+    or remove this abstract class in any future release. Custom implementations
+    must be updated to match interface changes.
+    """
 
     @abstractmethod
     async def connect(self) -> None:

src/claude_code_sdk/_internal/transport/subprocess_cli.py

@@ -39,7 +39,7 @@ def __init__(
         self._is_streaming = not isinstance(prompt, str)
         self._options = options
         self._cli_path = str(cli_path) if cli_path else self._find_cli()
-        self._cwd = str(options.cwd) if options.cwd else None
+        self._cwd: str | None = None
         self._process: Process | None = None
         self._stdout_stream: TextReceiveStream | None = None
         self._stderr_stream: TextReceiveStream | None = None

src/claude_code_sdk/query.py

@@ -5,13 +5,15 @@
 from typing import Any
 
 from ._internal.client import InternalClient
+from ._internal.transport import Transport
 from .types import ClaudeCodeOptions, Message
 
 
 async def query(
     *,
     prompt: str | AsyncIterable[dict[str, Any]],
     options: ClaudeCodeOptions | None = None,
+    transport: Transport | None = None,
 ) -> AsyncIterator[Message]:
     """
     Query Claude Code for one-shot or unidirectional streaming interactions.
@@ -56,6 +58,9 @@ async def query(
                  - 'acceptEdits': Auto-accept file edits
                  - 'bypassPermissions': Allow all tools (use with caution)
                  Set options.cwd for working directory.
+        transport: Optional transport implementation. If provided, this will be used
+                  instead of the default transport selection based on options.
+                  The transport will be automatically configured with the prompt and options.
 
     Yields:
         Messages from the conversation
@@ -90,6 +95,23 @@ async def prompts():
         async for message in query(prompt=prompts()):
             print(message)
         ```
+
+    Example - With custom transport:
+        ```python
+        from claude_code_sdk import query, Transport
+
+        class MyCustomTransport(Transport):
+            # Implement custom transport logic
+            pass
+
+        transport = MyCustomTransport()
+        async for message in query(
+            prompt="Hello",
+            transport=transport
+        ):
+            print(message)
+        ```
+
     """
     if options is None:
         options = ClaudeCodeOptions()
@@ -98,5 +120,7 @@ async def prompts():
 
     client = InternalClient()
 
-    async for message in client.process_query(prompt=prompt, options=options):
+    async for message in client.process_query(
+        prompt=prompt, options=options, transport=transport
+    ):
         yield message