Skip to main content

transports.acp

ACP (Agent Client Protocol) adapter.

Implements the Agent Client Protocol for agent-runtimes using the official Python SDK from https://github.com/agentclientprotocol/python-sdk

Protocol Reference: https://agentclientprotocol.com

Key Protocol Features:

  • JSON-RPC 2.0 based communication
  • Protocol version: 1 (integer for MAJOR version)
  • Methods: initialize, session/new, session/prompt, session/load, session/cancel
  • Session updates: session/update notifications with various update types
  • Identity context support for OAuth token propagation

ACPSession Objects​

@dataclass
class ACPSession()

An ACP session.

Attributes:

  • id - Unique session identifier.
  • created_at - Unix timestamp when created.
  • cwd - Current working directory.
  • context - Agent context for this session.
  • mcp_servers - Connected MCP servers.
  • current_mode - Current session mode.
  • cancelled - Whether the session has been cancelled.
  • identities - OAuth identities for this session (provider → token).
  • agent_id - Agent identifier for this session.
  • status - Session status (active, cancelled, completed).
  • created_at0 - Additional session metadata.

ACPTransport Objects​

class ACPTransport(BaseTransport)

Agent Client Protocol (ACP) adapter.

Implements the ACP protocol for agent-runtimes using official ACP SDK types and utilities for protocol compliance.

Protocol Reference: https://agentclientprotocol.com

The adapter supports:

  • initialize: Capability negotiation
  • session/new: Create new sessions
  • session/load: Load existing sessions
  • session/prompt: Run prompts with streaming updates
  • session/cancel: Cancel running prompts

Example:

agent = PydanticAIAgent(pydantic_agent) adapter = ACPTransport(agent)

Handle ACP initialize request

response = await adapter.handle_initialize(InitializeRequest(...))

Handle ACP session/prompt request

async for event in adapter.handle_prompt(PromptRequest(...)): yield event

__init__​

def __init__(agent: BaseAgent,
             agent_info: Optional[Implementation] = None,
             permission_handler: Optional[Callable[[dict[str, Any]],
                                                   bool]] = None)

Initialize the ACP adapter.

Arguments:

  • agent - The agent to adapt.
  • agent_info - Optional agent implementation info (name, version).
  • permission_handler - Optional callback for permission requests.

protocol_name​

@property
def protocol_name() -> str

Get the protocol name.

protocol_version​

@property
def protocol_version() -> int

Get the protocol version from official SDK.

capabilities​

@property
def capabilities() -> AgentCapabilities

Get agent capabilities.

handle_initialize​

async def handle_initialize(request: InitializeRequest) -> InitializeResponse

Handle ACP initialize request.

Per ACP spec, negotiates protocol version and exchanges capabilities.

Arguments:

  • request - Initialize request from client.

Returns:

Initialize response with protocol version and capabilities.

handle_new_session​

async def handle_new_session(
        request: NewSessionRequest,
        identities: Optional[list[dict[str,
                                       Any]]] = None) -> NewSessionResponse

Handle ACP session/new request.

Creates a new session with the specified configuration.

Arguments:

  • request - New session request from client.
  • identities - Optional OAuth identities (provider → token) for this session.

Returns:

New session response with session ID.

handle_load_session​

async def handle_load_session(
        request: LoadSessionRequest) -> Optional[LoadSessionResponse]

Handle ACP session/load request.

Loads an existing session by ID.

Arguments:

  • request - Load session request from client.

Returns:

Load session response or None if not supported/found.

handle_prompt​

async def handle_prompt(
    request: PromptRequest,
    identities: Optional[list[dict[str, Any]]] = None
) -> AsyncIterator[SessionNotification | PromptResponse]

Handle ACP session/prompt request.

Processes a prompt and yields session update notifications, followed by the final prompt response.

Arguments:

  • request - Prompt request from client.
  • identities - Optional OAuth identities to override session identities.

Yields:

Session notifications during processing, then PromptResponse.

handle_cancel​

async def handle_cancel(notification: CancelNotification) -> None

Handle ACP session/cancel notification.

Cancels ongoing operations for the specified session.

Arguments:

  • notification - Cancel notification from client.

handle_request​

async def handle_request(request: dict[str, Any]) -> dict[str, Any]

Handle a non-streaming ACP JSON-RPC request.

Arguments:

  • request - JSON-RPC request dictionary.

Returns:

JSON-RPC response dictionary.

handle_stream​

async def handle_stream(
        request: dict[str, Any]) -> AsyncIterator[dict[str, Any]]

Handle a streaming ACP JSON-RPC request.

Arguments:

  • request - JSON-RPC request dictionary.

Yields:

JSON-RPC notification or response dictionaries.

get_session​

def get_session(session_id: str) -> Optional[ACPSession]

Get a session by ID.

Arguments:

  • session_id - Session ID.

Returns:

Session or None.

active_session​

@property
def active_session() -> Optional[ACPSession]

Get the active session.