Skip to main content

routes.acp

ACP (Agent Client Protocol) routes for agent-runtimes server.

Provides WebSocket-based communication for agent interactions following the ACP specification from https://agentclientprotocol.com

Uses the official ACP Python SDK from: https://github.com/agentclientprotocol/python-sdk

Protocol Features:

  • JSON-RPC 2.0 message format
  • Protocol version: 1 (integer for MAJOR version)
  • Session-based communication
  • Streaming via session/update notifications

Supported Methods:

  • initialize: Capability negotiation
  • session/new: Create new session
  • session/load: Load existing session
  • session/prompt: Send prompt to agent
  • session/set_mode: Change session mode
  • session/cancel: Cancel running prompt

AgentCapabilities Objects​

class AgentCapabilities(BaseModel)

Agent capabilities for ACP protocol (extended info).

AgentInfo Objects​

class AgentInfo(BaseModel)

Agent information for ACP discovery.

SessionInfo Objects​

class SessionInfo(BaseModel)

Session information for ACP.

ACPMessage Objects​

class ACPMessage(BaseModel)

Base ACP message format.

Per JSON-RPC 2.0 spec, id can be a string, number, or null.

ACPError Objects​

class ACPError(BaseModel)

ACP error response.

register_prompt​

def register_prompt(session_id: str) -> asyncio.Event

Register a running prompt and return its cancellation event.

Arguments:

  • session_id - The session identifier.

Returns:

An asyncio.Event that can be set to signal cancellation.

unregister_prompt​

def unregister_prompt(session_id: str) -> None

Unregister a prompt when it completes.

Arguments:

  • session_id - The session identifier to remove.

cancel_prompt​

def cancel_prompt(session_id: str) -> bool

Cancel a running prompt and interrupt the sandbox.

Arguments:

  • session_id - The session identifier to cancel.

Returns:

True if the prompt was found and cancelled, False otherwise.

cancel_all_prompts​

def cancel_all_prompts() -> int

Cancel all running prompts and interrupt the sandbox.

Returns:

Number of prompts cancelled.

register_agent​

def register_agent(agent: BaseAgent, info: AgentInfo) -> None

Register an agent with the ACP server.

unregister_agent​

def unregister_agent(agent_id: str) -> None

Unregister an agent from the ACP server.

list_agents​

@router.get("/agents")
async def list_agents() -> dict[str, Any]

List all available agents.

Returns:

List of agent information.

get_agent​

@router.get("/agents/{agent_id:path}")
async def get_agent(agent_id: str) -> AgentInfo

Get information about a specific agent.

Arguments:

  • agent_id - The agent identifier.

Returns:

Agent information.

Raises:

  • HTTPException - If agent not found.

list_sessions​

@router.get("/sessions")
async def list_sessions() -> dict[str, Any]

List all active sessions.

Returns:

List of session information.

get_session​

@router.get("/sessions/{session_id}")
async def get_session(session_id: str) -> SessionInfo

Get information about a specific session.

Arguments:

  • session_id - The session identifier.

Returns:

Session information.

Raises:

  • HTTPException - If session not found.

close_session​

@router.delete("/sessions/{session_id}")
async def close_session(session_id: str) -> dict[str, str]

Close an active session.

Arguments:

  • session_id - The session identifier.

Returns:

Confirmation message.

Raises:

  • HTTPException - If session not found.

websocket_endpoint​

@router.websocket("/ws/{agent_id:path}")
async def websocket_endpoint(websocket: WebSocket, agent_id: str) -> None

WebSocket endpoint for ACP communication.

Implements the ACP protocol over WebSocket for real-time bidirectional communication with agents.

Arguments:

  • websocket - The WebSocket connection.
  • agent_id - The target agent identifier.

get_registered_agents​

def get_registered_agents() -> list[AgentInfo]

Get all registered agents.

get_active_sessions​

def get_active_sessions() -> list[SessionInfo]

Get all active sessions.