ACP API Reference¶

@dataclass
class JSONRPCError:
    """A JSON-RPC 2.0 error object.

    Attributes:
        code: Numeric error code.
        message: Human-readable error message.
        data: Optional additional error data.
    """

    code: int = INTERNAL_ERROR
    message: str = "Internal error"
    data: Any = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize to a dictionary."""
        d: dict[str, Any] = {"code": self.code, "message": self.message}
        if self.data is not None:
            d["data"] = self.data
        return d

    @classmethod
    def from_dict(cls, raw: dict[str, Any]) -> JSONRPCError:
        """Deserialize from a dictionary."""
        return cls(
            code=raw["code"],
            message=raw["message"],
            data=raw.get("data"),
        )

class JSONRPCException(Exception):
    """Exception wrapper around a ``JSONRPCError`` data object.

    Attributes:
        error: The underlying ``JSONRPCError`` instance.
    """

    def __init__(self, error: JSONRPCError) -> None:
        super().__init__(error.message)
        self.error = error

class JSONRPCTransport:
    """Async JSON-RPC 2.0 transport over newline-delimited JSON streams.

    Each message is a single JSON object terminated by ``\\n``.
    Messages must not contain embedded newlines.

    Attributes:
        reader: Async stream to read incoming messages from.
        writer: Async stream to write outgoing messages to.
    """

    def __init__(
        self,
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
    ) -> None:
        self.reader = reader
        self.writer = writer
        self._closed = False

    async def read_message(self) -> dict[str, Any] | None:
        """Read the next JSON-RPC message.

        Returns:
            Parsed JSON object, or ``None`` on EOF.
        """
        while True:
            line = await self.reader.readline()
            if not line:
                return None
            text = line.decode("utf-8").strip()
            if not text:
                continue
            try:
                return json.loads(text)
            except json.JSONDecodeError:
                logger.warning("Ignoring malformed JSON line: %s", text[:120])

    async def write_message(self, msg: dict[str, Any]) -> None:
        """Write a JSON-RPC message followed by a newline.

        Args:
            msg: JSON-serializable dictionary to send.
        """
        raw = json.dumps(msg, separators=(",", ":"), ensure_ascii=False)
        self.writer.write((raw + "\n").encode("utf-8"))
        await self.writer.drain()

    async def send_request(
        self,
        method: str,
        params: Any = None,
        req_id: Union[int, str, None] = None,
    ) -> dict[str, Any]:
        """Build and send a JSON-RPC request.

        Args:
            method: The RPC method name.
            params: Parameters for the method.
            req_id: Optional explicit request id.

        Returns:
            The message dictionary that was sent.
        """
        msg: dict[str, Any] = {
            "jsonrpc": JSONRPC_VERSION,
            "id": req_id if req_id is not None else next_id(),
            "method": method,
        }
        if params is not None:
            msg["params"] = params
        await self.write_message(msg)
        return msg

    async def send_notification(self, method: str, params: Any = None) -> None:
        """Build and send a JSON-RPC notification (no ``id``).

        Args:
            method: The RPC method name.
            params: Parameters for the notification.
        """
        msg: dict[str, Any] = {"jsonrpc": JSONRPC_VERSION, "method": method}
        if params is not None:
            msg["params"] = params
        await self.write_message(msg)

    async def send_result(self, req_id: Union[int, str], result: Any) -> None:
        """Send a JSON-RPC success response.

        Args:
            req_id: The id of the original request.
            result: The result payload.
        """
        await self.write_message(
            {"jsonrpc": JSONRPC_VERSION, "id": req_id, "result": result}
        )

    async def send_error(
        self, req_id: Union[int, str, None], error: JSONRPCError
    ) -> None:
        """Send a JSON-RPC error response.

        Args:
            req_id: The id of the original request (may be ``None``).
            error: The error object.
        """
        await self.write_message(
            {"jsonrpc": JSONRPC_VERSION, "id": req_id, "error": error.to_dict()}
        )

    async def close(self) -> None:
        """Close the writer stream."""
        if not self._closed:
            self._closed = True
            self.writer.close()

    @property
    def is_closed(self) -> bool:
        """Whether the writer has been closed."""
        return self._closed

class StopReason(str, Enum):
    """Reason an agent stopped a prompt turn."""

    END_TURN = "end_turn"
    MAX_TOKENS = "max_tokens"
    MAX_TURN_REQUESTS = "max_turn_requests"
    REFUSAL = "refusal"
    CANCELLED = "cancelled"

class ToolKind(str, Enum):
    """Category of a tool being invoked."""

    READ = "read"
    EDIT = "edit"
    DELETE = "delete"
    MOVE = "move"
    SEARCH = "search"
    EXECUTE = "execute"
    THINK = "think"
    FETCH = "fetch"
    OTHER = "other"

class ToolCallStatus(str, Enum):
    """Execution status of a tool call."""

    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"

class PermissionOptionKind(str, Enum):
    """Kind of permission option presented to the user."""

    ALLOW_ONCE = "allow_once"
    ALLOW_ALWAYS = "allow_always"
    REJECT_ONCE = "reject_once"
    REJECT_ALWAYS = "reject_always"

class PlanEntryPriority(str, Enum):
    """Priority of a plan entry."""

    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

class PlanEntryStatus(str, Enum):
    """Execution status of a plan entry."""

    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"

@dataclass
class TextContent:
    """Plain text content block.

    Attributes:
        text: The text payload.
        type: Discriminator (always ``"text"``).
    """

    text: str
    type: str = "text"

@dataclass
class ImageContent:
    """Base64-encoded image content block.

    Attributes:
        data: Base64-encoded image data.
        mime_type: MIME type such as ``"image/png"``.
        type: Discriminator (always ``"image"``).
    """

    data: str
    mime_type: str
    type: str = "image"

@dataclass
class AudioContent:
    """Base64-encoded audio content block.

    Attributes:
        data: Base64-encoded audio data.
        mime_type: MIME type such as ``"audio/wav"``.
        type: Discriminator (always ``"audio"``).
    """

    data: str
    mime_type: str
    type: str = "audio"

@dataclass
class ResourceContent:
    """Embedded resource content block.

    Attributes:
        resource: The embedded resource (text or blob).
        type: Discriminator (always ``"resource"``).
    """

    resource: _TextResource
    type: str = "resource"

@dataclass
class ResourceLinkContent:
    """Reference to an external resource.

    Attributes:
        uri: URI of the resource.
        name: Human-readable resource name.
        mime_type: Optional MIME type.
        title: Optional display title.
        description: Optional description.
        size: Optional size in bytes.
        type: Discriminator (always ``"resource_link"``).
    """

    uri: str
    name: str
    mime_type: str | None = None
    title: str | None = None
    description: str | None = None
    size: int | None = None
    type: str = "resource_link"

@dataclass
class ImplementationInfo:
    """Information about a client or agent implementation.

    Attributes:
        name: Programmatic identifier.
        version: Version string.
        title: Human-readable display name.
    """

    name: str
    version: str = ""
    title: str | None = None

@dataclass
class FsCapabilities:
    """Client file-system capabilities.

    Attributes:
        read_text_file: Whether ``fs/read_text_file`` is available.
        write_text_file: Whether ``fs/write_text_file`` is available.
    """

    read_text_file: bool = False
    write_text_file: bool = False

@dataclass
class ClientCapabilities:
    """Capabilities supported by the client.

    Attributes:
        fs: File-system method availability.
        terminal: Whether all ``terminal/*`` methods are available.
    """

    fs: FsCapabilities | None = None
    terminal: bool = False

@dataclass
class PromptCapabilities:
    """Content types the agent supports in prompts.

    Attributes:
        image: Whether image content is supported.
        audio: Whether audio content is supported.
        embedded_context: Whether embedded resource content is supported.
    """

    image: bool = False
    audio: bool = False
    embedded_context: bool = False

@dataclass
class McpCapabilities:
    """MCP transport capabilities.

    Attributes:
        http: Whether HTTP MCP transport is supported.
        sse: Whether SSE MCP transport is supported.
    """

    http: bool = False
    sse: bool = False

@dataclass
class SessionListCapability:
    """Marker for session/list support."""

    pass

@dataclass
class SessionCapabilities:
    """Session-level capabilities.

    Attributes:
        list: If present, ``session/list`` is supported.
    """

    list: SessionListCapability | None = None

@dataclass
class AgentCapabilities:
    """Capabilities supported by the agent.

    Attributes:
        load_session: Whether ``session/load`` is available.
        prompt_capabilities: Supported content types in prompts.
        mcp_capabilities: Supported MCP transports.
        session_capabilities: Session-level capabilities.
    """

    load_session: bool = False
    prompt_capabilities: PromptCapabilities | None = None
    mcp_capabilities: McpCapabilities | None = None
    session_capabilities: SessionCapabilities | None = None

@dataclass
class AuthMethod:
    """An authentication method advertised by the agent.

    Attributes:
        id: Unique identifier for this auth method.
        name: Human-readable name.
        description: Optional description.
    """

    id: str
    name: str
    description: str | None = None

@dataclass
class InitializeParams:
    """Parameters for the ``initialize`` request.

    Attributes:
        protocol_version: Latest protocol version the client supports.
        client_capabilities: Capabilities the client supports.
        client_info: Information about the client implementation.
    """

    protocol_version: int = 1
    client_capabilities: ClientCapabilities | None = None
    client_info: ImplementationInfo | None = None

@dataclass
class InitializeResult:
    """Result of the ``initialize`` request.

    Attributes:
        protocol_version: Negotiated protocol version.
        agent_capabilities: Capabilities the agent supports.
        agent_info: Information about the agent implementation.
        auth_methods: Available authentication methods.
    """

    protocol_version: int = 1
    agent_capabilities: AgentCapabilities | None = None
    agent_info: ImplementationInfo | None = None
    auth_methods: list[AuthMethod] | None = None

@dataclass
class EnvVariable:
    """An environment variable.

    Attributes:
        name: Variable name.
        value: Variable value.
    """

    name: str
    value: str

@dataclass
class McpServerStdio:
    """Stdio MCP server specification.

    Attributes:
        name: Human-readable server name.
        command: Path to the MCP server executable.
        args: Command-line arguments.
        env: Environment variables.
    """

    name: str
    command: str
    args: list[str] = field(default_factory=list)
    env: list[EnvVariable] | None = None

@dataclass
class HttpHeader:
    """An HTTP header.

    Attributes:
        name: Header name.
        value: Header value.
    """

    name: str
    value: str

@dataclass
class McpServerHttp:
    """HTTP MCP server specification.

    Attributes:
        name: Human-readable server name.
        url: URL of the MCP server.
        headers: HTTP headers.
        type: Transport type (``"http"`` or ``"sse"``).
    """

    name: str
    url: str
    headers: list[HttpHeader] = field(default_factory=list)
    type: str = "http"

@dataclass
class NewSessionParams:
    """Parameters for ``session/new``.

    Attributes:
        cwd: Absolute path to the working directory.
        mcp_servers: MCP servers to connect to.
    """

    cwd: str
    mcp_servers: list[dict[str, Any]] | None = None

@dataclass
class SessionMode:
    """An operating mode for the agent.

    Attributes:
        id: Unique mode identifier.
        name: Human-readable mode name.
        description: Optional description.
    """

    id: str
    name: str
    description: str | None = None

@dataclass
class SessionModeState:
    """Current mode state for a session.

    Attributes:
        current_mode_id: The currently active mode.
        available_modes: All available modes.
    """

    current_mode_id: str
    available_modes: list[SessionMode] = field(default_factory=list)

@dataclass
class ConfigOptionValue:
    """A possible value for a configuration option.

    Attributes:
        value: Value identifier.
        name: Human-readable name.
        description: Optional description.
    """

    value: str
    name: str
    description: str | None = None

@dataclass
class ConfigOption:
    """A session configuration option.

    Attributes:
        id: Unique option identifier.
        name: Human-readable label.
        type: Input control type (currently only ``"select"``).
        current_value: Currently selected value.
        options: Available values.
        description: Optional description.
        category: Optional semantic category.
    """

    id: str
    name: str
    type: str
    current_value: str
    options: list[ConfigOptionValue] = field(default_factory=list)
    description: str | None = None
    category: str | None = None

@dataclass
class NewSessionResult:
    """Result of ``session/new``.

    Attributes:
        session_id: Unique identifier for the created session.
        modes: Optional mode state.
        config_options: Optional configuration options.
    """

    session_id: str
    modes: SessionModeState | None = None
    config_options: list[ConfigOption] | None = None

@dataclass
class LoadSessionParams:
    """Parameters for ``session/load``.

    Attributes:
        session_id: Session to resume.
        cwd: Working directory.
        mcp_servers: MCP servers to connect to.
    """

    session_id: str
    cwd: str
    mcp_servers: list[dict[str, Any]] | None = None

@dataclass
class PromptParams:
    """Parameters for ``session/prompt``.

    Attributes:
        session_id: Target session.
        prompt: Content blocks forming the user message.
    """

    session_id: str
    prompt: list[ContentBlock] = field(default_factory=list)

@dataclass
class PromptResult:
    """Result of ``session/prompt``.

    Attributes:
        stop_reason: Why the agent stopped.
    """

    stop_reason: StopReason

@dataclass
class CancelParams:
    """Parameters for ``session/cancel`` notification.

    Attributes:
        session_id: Session to cancel.
    """

    session_id: str

@dataclass
class SetModeParams:
    """Parameters for ``session/set_mode``.

    Attributes:
        session_id: Target session.
        mode_id: Mode to switch to.
    """

    session_id: str
    mode_id: str

@dataclass
class SetConfigOptionParams:
    """Parameters for ``session/set_config_option``.

    Attributes:
        session_id: Target session.
        config_id: Configuration option id.
        value: New value.
    """

    session_id: str
    config_id: str
    value: str

@dataclass
class SetConfigOptionResult:
    """Result of ``session/set_config_option``.

    Attributes:
        config_options: Complete list of config options with current values.
    """

    config_options: list[ConfigOption] = field(default_factory=list)

@dataclass
class ListSessionsParams:
    """Parameters for ``session/list``.

    Attributes:
        cwd: Optional directory filter.
        cursor: Optional pagination cursor.
    """

    cwd: str | None = None
    cursor: str | None = None

@dataclass
class SessionInfo:
    """Metadata about an existing session.

    Attributes:
        session_id: Unique session identifier.
        cwd: Working directory.
        title: Optional human-readable title.
        updated_at: Optional ISO 8601 timestamp.
    """

    session_id: str
    cwd: str
    title: str | None = None
    updated_at: str | None = None

@dataclass
class ListSessionsResult:
    """Result of ``session/list``.

    Attributes:
        sessions: List of session metadata.
        next_cursor: Pagination cursor for the next page.
    """

    sessions: list[SessionInfo] = field(default_factory=list)
    next_cursor: str | None = None

@dataclass
class ToolCallLocation:
    """File location affected by a tool call.

    Attributes:
        path: Absolute file path.
        line: Optional line number (1-based).
    """

    path: str
    line: int | None = None

@dataclass
class DiffContent:
    """A file diff produced by a tool call.

    Attributes:
        path: Absolute path of the file being modified.
        new_text: New content after modification.
        old_text: Original content (``None`` for new files).
        type: Discriminator (always ``"diff"``).
    """

    path: str
    new_text: str
    old_text: str | None = None
    type: str = "diff"