Transport Layer

class Transport(ABC):
    """
    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
    remove this abstract class in any future release. Custom implementations
    must be updated to match interface changes.

    This is a low-level transport interface that handles raw I/O with the Claude
    process or service. The Query class builds on top of this to implement the
    control protocol and message routing.

    Use cases for custom transports:
    - Connect to remote Claude Code instances
    - Implement custom communication protocols
    - Add middleware (logging, encryption, compression)
    - Support alternative execution environments

    Most users should use the default subprocess transport via query() or
    ClaudeSDKClient rather than implementing custom transports.
    """

    @abstractmethod
    async def connect(self) -> None:
        """
        Connect the transport and prepare for communication.

        For subprocess transports, this starts the process.
        For network transports, this establishes the connection.

        This method must be called before any read or write operations.
        Should be idempotent - calling multiple times should be safe.

        Raises:
            CLIConnectionError: If connection fails
            CLINotFoundError: If Claude CLI is not found (subprocess transport)

        Example:
            transport = MyTransport()
            await transport.connect()
        """

    @abstractmethod
    async def write(self, data: str) -> None:
        """
        Write raw data to the transport.

        Args:
            data: Raw string data to write (typically JSON + newline)

        The data should be a complete message, usually JSON-encoded with a
        trailing newline. The transport is responsible for delivering this
        data to the Claude process.

        Raises:
            Exception: If write fails or transport is not connected

        Example:
            await transport.write('{"type": "user", "content": "Hello"}\\n')
        """

    @abstractmethod
    def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        """
        Read and parse messages from the transport.

        This method returns an async iterator that yields parsed JSON messages
        from the Claude process. Each message is a dictionary.

        The implementation should:
        - Read lines from the transport
        - Parse each line as JSON
        - Yield the parsed message dictionary
        - Continue until the transport is closed

        Yields:
            Parsed JSON messages as dictionaries

        Raises:
            CLIJSONDecodeError: If JSON parsing fails
            Exception: If read fails or transport is not connected

        Example:
            async for message in transport.read_messages():
                print(f"Received: {message}")
        """

    @abstractmethod
    async def close(self) -> None:
        """
        Close the transport connection and clean up resources.

        This method should:
        - Close any open connections or file handles
        - Terminate any associated processes (subprocess transport)
        - Release any allocated resources
        - Be safe to call multiple times

        Should be idempotent - calling multiple times should be safe.

        Example:
            await transport.close()
        """

    @abstractmethod
    def is_ready(self) -> bool:
        """
        Check if transport is ready for communication.

        Returns:
            True if transport is ready to send/receive messages, False otherwise

        This method should return True after connect() succeeds and before
        close() is called. It's used to verify the transport is in a usable
        state.

        Example:
            if transport.is_ready():
                await transport.write(data)
        """

    @abstractmethod
    async def end_input(self) -> None:
        """
        End the input stream (close stdin for process transports).

        This signals to the Claude process that no more input will be sent.
        For subprocess transports, this closes stdin. For network transports,
        this might send an end-of-stream signal.

        After calling this, you should only read remaining output, not write
        new messages.

        Example:
            await transport.end_input()
            # Can still read remaining messages
            async for msg in transport.read_messages():
                print(msg)
        """

from claude_agent_sdk import query

# Most users should just use query() which handles transport internally
async for msg in query(prompt="Hello"):
    print(msg)

# Or use ClaudeSDKClient which also manages transport
from claude_agent_sdk import ClaudeSDKClient

async with ClaudeSDKClient() as client:
    await client.query("Hello")
    async for msg in client.receive_response():
        print(msg)

from claude_agent_sdk import Transport
from typing import AsyncIterator, Any

class MyCustomTransport(Transport):
    """Custom transport implementation."""

    def __init__(self):
        self._connected = False
        self._reader = None
        self._writer = None

    async def connect(self) -> None:
        """Connect to Claude."""
        if self._connected:
            return

        # Implement connection logic
        # For network: establish TCP/HTTP connection
        # For subprocess: start process
        # Set self._reader and self._writer

        self._connected = True

    async def write(self, data: str) -> None:
        """Write data to Claude."""
        if not self._connected:
            raise Exception("Not connected")

        # Implement write logic
        # For network: send over socket
        # For subprocess: write to stdin
        await self._writer.write(data.encode())

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        """Read messages from Claude."""
        if not self._connected:
            raise Exception("Not connected")

        # Implement read logic
        while True:
            line = await self._reader.readline()
            if not line:
                break

            # Parse and yield message
            import json
            try:
                message = json.loads(line.decode())
                yield message
            except json.JSONDecodeError as e:
                from claude_agent_sdk import CLIJSONDecodeError
                raise CLIJSONDecodeError(line.decode(), e)

    async def close(self) -> None:
        """Close the transport."""
        if not self._connected:
            return

        # Implement close logic
        if self._writer:
            self._writer.close()
            await self._writer.wait_closed()

        self._connected = False

    def is_ready(self) -> bool:
        """Check if ready."""
        return self._connected

    async def end_input(self) -> None:
        """End input stream."""
        if self._writer:
            self._writer.write_eof()

import asyncio
import json
from claude_agent_sdk import Transport, ClaudeSDKClient
from typing import AsyncIterator, Any

class NetworkTransport(Transport):
    """Transport for remote Claude Code over network."""

    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        self._reader = None
        self._writer = None

    async def connect(self) -> None:
        """Connect to remote Claude Code."""
        self._reader, self._writer = await asyncio.open_connection(
            self.host,
            self.port
        )

    async def write(self, data: str) -> None:
        """Write to network socket."""
        self._writer.write(data.encode('utf-8'))
        await self._writer.drain()

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        """Read from network socket."""
        while True:
            line = await self._reader.readline()
            if not line:
                break

            try:
                message = json.loads(line.decode('utf-8'))
                yield message
            except json.JSONDecodeError as e:
                from claude_agent_sdk import CLIJSONDecodeError
                raise CLIJSONDecodeError(line.decode(), e)

    async def close(self) -> None:
        """Close network connection."""
        if self._writer:
            self._writer.close()
            await self._writer.wait_closed()

    def is_ready(self) -> bool:
        """Check if connected."""
        return self._writer is not None and not self._writer.is_closing()

    async def end_input(self) -> None:
        """Close write side of connection."""
        if self._writer:
            self._writer.write_eof()
            await self._writer.drain()

# Use custom transport
transport = NetworkTransport("claude.example.com", 8080)
client = ClaudeSDKClient(transport=transport)

async with client:
    await client.query("Hello")
    async for msg in client.receive_response():
        print(msg)

import logging
from claude_agent_sdk import Transport, query
from typing import AsyncIterator, Any

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

class LoggingTransport(Transport):
    """Transport wrapper that logs all I/O."""

    def __init__(self, wrapped: Transport):
        self._transport = wrapped

    async def connect(self) -> None:
        logger.info("Connecting transport")
        await self._transport.connect()
        logger.info("Transport connected")

    async def write(self, data: str) -> None:
        logger.debug(f"Write: {data[:100]}...")
        await self._transport.write(data)

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        logger.info("Starting to read messages")
        async for message in self._transport.read_messages():
            logger.debug(f"Read: {message}")
            yield message
        logger.info("Finished reading messages")

    async def close(self) -> None:
        logger.info("Closing transport")
        await self._transport.close()
        logger.info("Transport closed")

    def is_ready(self) -> bool:
        ready = self._transport.is_ready()
        logger.debug(f"Transport ready: {ready}")
        return ready

    async def end_input(self) -> None:
        logger.info("Ending input")
        await self._transport.end_input()

# Use with default transport wrapped in logging
from claude_agent_sdk._internal.transport.subprocess_cli import SubprocessTransport

base_transport = SubprocessTransport()
logged_transport = LoggingTransport(base_transport)

async for msg in query(prompt="Hello", transport=logged_transport):
    print(msg)

import anyio
from claude_agent_sdk import Transport, CLIConnectionError
from typing import AsyncIterator, Any

class RetryingTransport(Transport):
    """Transport wrapper with automatic retry."""

    def __init__(self, wrapped: Transport, max_retries: int = 3):
        self._transport = wrapped
        self._max_retries = max_retries

    async def connect(self) -> None:
        """Connect with retry."""
        for attempt in range(self._max_retries):
            try:
                await self._transport.connect()
                return
            except CLIConnectionError as e:
                if attempt < self._max_retries - 1:
                    print(f"Connection failed, retrying... ({attempt + 1}/{self._max_retries})")
                    await anyio.sleep(2 ** attempt)  # Exponential backoff
                else:
                    raise

    async def write(self, data: str) -> None:
        """Write with retry."""
        for attempt in range(self._max_retries):
            try:
                await self._transport.write(data)
                return
            except Exception as e:
                if attempt < self._max_retries - 1:
                    print(f"Write failed, retrying... ({attempt + 1}/{self._max_retries})")
                    await anyio.sleep(1)
                else:
                    raise

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        """Delegate to wrapped transport."""
        async for message in self._transport.read_messages():
            yield message

    async def close(self) -> None:
        """Delegate to wrapped transport."""
        await self._transport.close()

    def is_ready(self) -> bool:
        """Delegate to wrapped transport."""
        return self._transport.is_ready()

    async def end_input(self) -> None:
        """Delegate to wrapped transport."""
        await self._transport.end_input()

import time
from claude_agent_sdk import Transport
from typing import AsyncIterator, Any

class MetricsTransport(Transport):
    """Transport wrapper that collects metrics."""

    def __init__(self, wrapped: Transport):
        self._transport = wrapped
        self.bytes_written = 0
        self.bytes_read = 0
        self.messages_read = 0
        self.connect_time = 0

    async def connect(self) -> None:
        start = time.time()
        await self._transport.connect()
        self.connect_time = time.time() - start

    async def write(self, data: str) -> None:
        self.bytes_written += len(data)
        await self._transport.write(data)

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        async for message in self._transport.read_messages():
            self.messages_read += 1
            # Estimate bytes (rough)
            self.bytes_read += len(str(message))
            yield message

    async def close(self) -> None:
        await self._transport.close()

    def is_ready(self) -> bool:
        return self._transport.is_ready()

    async def end_input(self) -> None:
        await self._transport.end_input()

    def print_metrics(self):
        print(f"Connect time: {self.connect_time:.3f}s")
        print(f"Bytes written: {self.bytes_written}")
        print(f"Bytes read: {self.bytes_read}")
        print(f"Messages read: {self.messages_read}")

# Usage
from claude_agent_sdk._internal.transport.subprocess_cli import SubprocessTransport

base_transport = SubprocessTransport()
metrics_transport = MetricsTransport(base_transport)

async for msg in query(prompt="Hello", transport=metrics_transport):
    print(msg)

metrics_transport.print_metrics()

from enum import Enum
from claude_agent_sdk import Transport
from typing import AsyncIterator, Any

class TransportState(Enum):
    DISCONNECTED = "disconnected"
    CONNECTING = "connecting"
    CONNECTED = "connected"
    CLOSING = "closing"
    CLOSED = "closed"

class StatefulTransport(Transport):
    """Transport with explicit state management."""

    def __init__(self, wrapped: Transport):
        self._transport = wrapped
        self._state = TransportState.DISCONNECTED

    async def connect(self) -> None:
        if self._state == TransportState.CONNECTED:
            return

        self._state = TransportState.CONNECTING
        try:
            await self._transport.connect()
            self._state = TransportState.CONNECTED
        except Exception:
            self._state = TransportState.DISCONNECTED
            raise

    async def write(self, data: str) -> None:
        if self._state != TransportState.CONNECTED:
            raise Exception(f"Cannot write in state: {self._state}")
        await self._transport.write(data)

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        if self._state != TransportState.CONNECTED:
            raise Exception(f"Cannot read in state: {self._state}")
        async for message in self._transport.read_messages():
            yield message

    async def close(self) -> None:
        if self._state in (TransportState.CLOSING, TransportState.CLOSED):
            return

        self._state = TransportState.CLOSING
        try:
            await self._transport.close()
        finally:
            self._state = TransportState.CLOSED

    def is_ready(self) -> bool:
        return self._state == TransportState.CONNECTED

    async def end_input(self) -> None:
        if self._state == TransportState.CONNECTED:
            await self._transport.end_input()

    @property
    def state(self) -> TransportState:
        return self._state

import json
from claude_agent_sdk._internal.transport.subprocess_cli import SubprocessTransport
from claude_agent_sdk import ClaudeAgentOptions

async def use_transport_directly():
    """Example of using transport directly (advanced)."""
    # Create and connect transport
    transport = SubprocessTransport(options=ClaudeAgentOptions())
    await transport.connect()

    try:
        # Send a message
        message = {
            "type": "user",
            "message": {"role": "user", "content": "Hello"},
            "session_id": "default"
        }
        await transport.write(json.dumps(message) + "\n")

        # Read responses
        async for response in transport.read_messages():
            print(f"Response: {response}")

            # Check for completion
            if response.get("type") == "result":
                break

    finally:
        # Clean up
        await transport.end_input()
        await transport.close()

# Note: Most users should use query() or ClaudeSDKClient instead
await use_transport_directly()

from claude_agent_sdk import Transport
from typing import AsyncIterator, Any
import json

class MockTransport(Transport):
    """Mock transport for testing."""

    def __init__(self, responses: list[dict]):
        self.responses = responses
        self.written = []
        self._connected = False
        self._response_index = 0

    async def connect(self) -> None:
        self._connected = True

    async def write(self, data: str) -> None:
        self.written.append(json.loads(data))

    async def read_messages(self) -> AsyncIterator[dict[str, Any]]:
        for response in self.responses:
            yield response

    async def close(self) -> None:
        self._connected = False

    def is_ready(self) -> bool:
        return self._connected

    async def end_input(self) -> None:
        pass

# Use in tests
async def test_query():
    mock = MockTransport([
        {"type": "assistant", "content": [{"type": "text", "text": "Hello!"}]},
        {"type": "result", "is_error": False}
    ])

    # Test your code with mock transport
    # ...