ant_ai.acp.server

ACPServer pydantic-model

Bases: BaseModel

Serve an ant-ai agent over the Agent Client Protocol (ACP).

Supports two modes:

• ASGI / WebSocket – call :meth:starlette_app or :meth:fastapi_app to get an ASGI application that exposes the agent at /acp/ws.
• stdio – call :meth:serve_stdio to run the agent as a stdio ACP process that editors such as Zed or Gemini CLI can spawn directly.

To serve both A2A and ACP from a single process, combine the routes from each server's ASGI app — they occupy disjoint paths and compose cleanly:

    from starlette.applications import Starlette

    a2a = A2AServer(agent=agent, workflow=wf, agent_card=card)
    acp = ACPServer(agent=agent, workflow=wf)

    app = Starlette(routes=[*a2a.starlette_app().routes, *acp.starlette_app().routes])
    # A2A: POST /  and GET /.well-known/agent-card.json
    # ACP: WS  /acp/ws

Config:

• arbitrary_types_allowed: True

Fields:

• agent (Agent)
• workflow (Workflow)
• host (str)
• port (int)
• commands (list[ACPCommand])
• context_class (type[InvocationContext])

Source code in src/ant_ai/acp/server.py

class ACPServer(BaseModel):
    """Serve an ant-ai agent over the Agent Client Protocol (ACP).

    Supports two modes:

    - **ASGI / WebSocket** – call :meth:`starlette_app` or :meth:`fastapi_app` to get an
      ASGI application that exposes the agent at ``/acp/ws``.
    - **stdio** – call :meth:`serve_stdio` to run the agent as a stdio ACP process that
      editors such as Zed or Gemini CLI can spawn directly.

    To serve both A2A and ACP from a single process, combine the routes from each server's
    ASGI app — they occupy disjoint paths and compose cleanly:

    ```python
        from starlette.applications import Starlette

        a2a = A2AServer(agent=agent, workflow=wf, agent_card=card)
        acp = ACPServer(agent=agent, workflow=wf)

        app = Starlette(routes=[*a2a.starlette_app().routes, *acp.starlette_app().routes])
        # A2A: POST /  and GET /.well-known/agent-card.json
        # ACP: WS  /acp/ws
    ```
    """

    agent: Annotated[Agent, SkipValidation]
    workflow: Annotated[Workflow, SkipValidation]
    host: str = Field(default="127.0.0.1")
    port: int = Field(default=9001)
    commands: Annotated[list[ACPCommand], SkipValidation] = Field(default_factory=list)
    context_class: type[InvocationContext] = Field(
        default=InvocationContext,
        description=(
            "The InvocationContext (sub)class built for each prompt. Subclass "
            "InvocationContext to carry your own fields through the run."
        ),
    )

    model_config = ConfigDict(arbitrary_types_allowed=True)

    def build_routes(self) -> list:
        return [
            build_acp_ws_route(
                self.agent,
                self.workflow,
                commands=self.commands or None,
                context_class=self.context_class,
            )
        ]

    def starlette_app(self) -> Starlette:
        """Create a Starlette application serving ACP over WebSocket."""
        return Starlette(routes=self.build_routes())

    def fastapi_app(self) -> FastAPI:
        """Create a FastAPI application serving ACP over WebSocket."""
        return FastAPI(title=self.agent.name, routes=self.build_routes())

    def serve(self) -> None:
        """Start a uvicorn server exposing ACP over WebSocket at ``/acp/ws``."""
        try:
            import uvicorn

            logger.info(
                f"Starting ACP WebSocket server for agent '{self.agent.name}' "
                f"at {self.host}:{self.port} (ws://{self.host}:{self.port}/acp/ws)..."
            )
            uvicorn.run(self.starlette_app(), host=self.host, port=self.port)
        except ImportError as e:
            raise ImportError(
                "Uvicorn is not installed. Please install it with 'uv add uvicorn'."
            ) from e
        except KeyboardInterrupt:
            logger.info("Server stopped")
        except Exception as e:
            logger.error(f"Failed to start server: {e}")
            raise RuntimeError(f"Failed to start the server: {e}") from e

    def serve_stdio(self) -> None:
        """Run the agent as a stdio ACP process (for editors/CLIs that spawn agents)."""
        import asyncio

        from acp import run_agent

        logger.info(f"Starting ACP stdio agent '{self.agent.name}'...")
        adapter = ACPAdapter(
            self.agent,
            self.workflow,
            commands=self.commands or None,
            context_class=self.context_class,
        )
        asyncio.run(run_agent(adapter, use_unstable_protocol=True))

context_class pydantic-field

context_class: type[InvocationContext] = InvocationContext

The InvocationContext (sub)class built for each prompt. Subclass InvocationContext to carry your own fields through the run.

starlette_app

starlette_app() -> Starlette

Create a Starlette application serving ACP over WebSocket.

Source code in src/ant_ai/acp/server.py

def starlette_app(self) -> Starlette:
    """Create a Starlette application serving ACP over WebSocket."""
    return Starlette(routes=self.build_routes())

fastapi_app

fastapi_app() -> FastAPI

Create a FastAPI application serving ACP over WebSocket.

Source code in src/ant_ai/acp/server.py

def fastapi_app(self) -> FastAPI:
    """Create a FastAPI application serving ACP over WebSocket."""
    return FastAPI(title=self.agent.name, routes=self.build_routes())

serve

serve() -> None

Start a uvicorn server exposing ACP over WebSocket at /acp/ws.

Source code in src/ant_ai/acp/server.py

def serve(self) -> None:
    """Start a uvicorn server exposing ACP over WebSocket at ``/acp/ws``."""
    try:
        import uvicorn

        logger.info(
            f"Starting ACP WebSocket server for agent '{self.agent.name}' "
            f"at {self.host}:{self.port} (ws://{self.host}:{self.port}/acp/ws)..."
        )
        uvicorn.run(self.starlette_app(), host=self.host, port=self.port)
    except ImportError as e:
        raise ImportError(
            "Uvicorn is not installed. Please install it with 'uv add uvicorn'."
        ) from e
    except KeyboardInterrupt:
        logger.info("Server stopped")
    except Exception as e:
        logger.error(f"Failed to start server: {e}")
        raise RuntimeError(f"Failed to start the server: {e}") from e

serve_stdio

serve_stdio() -> None

Run the agent as a stdio ACP process (for editors/CLIs that spawn agents).

Source code in src/ant_ai/acp/server.py

def serve_stdio(self) -> None:
    """Run the agent as a stdio ACP process (for editors/CLIs that spawn agents)."""
    import asyncio

    from acp import run_agent

    logger.info(f"Starting ACP stdio agent '{self.agent.name}'...")
    adapter = ACPAdapter(
        self.agent,
        self.workflow,
        commands=self.commands or None,
        context_class=self.context_class,
    )
    asyncio.run(run_agent(adapter, use_unstable_protocol=True))

build_acp_ws_route

build_acp_ws_route(
    agent: Agent,
    workflow: Workflow,
    *,
    commands: list[ACPCommand] | None = None,
    context_class: type[
        InvocationContext
    ] = InvocationContext,
) -> WebSocketRoute

Return a Starlette WebSocketRoute that bridges ACP over WebSocket at /acp/ws.

Source code in src/ant_ai/acp/server.py

def build_acp_ws_route(
    agent: Agent,
    workflow: Workflow,
    *,
    commands: list[ACPCommand] | None = None,
    context_class: type[InvocationContext] = InvocationContext,
) -> WebSocketRoute:
    """Return a Starlette WebSocketRoute that bridges ACP over WebSocket at ``/acp/ws``."""
    adapter = ACPAdapter(
        agent, workflow, commands=commands, context_class=context_class
    )

    async def _handle_ws(websocket: WebSocket) -> None:
        from acp.agent.connection import AgentSideConnection

        await websocket.accept()

        sock_agent, sock_bridge = socket.socketpair()
        try:
            reader_agent, writer_agent = await asyncio.open_connection(sock=sock_agent)
            reader_bridge, writer_bridge = await asyncio.open_connection(
                sock=sock_bridge
            )
        except Exception:
            sock_agent.close()
            sock_bridge.close()
            await websocket.close()
            return

        # ``session/close``, ``session/fork`` and ``session/resume`` are gated behind the unstable protocol flag in acp; the adapter implements all three, so opt in to make them reachable.
        conn = AgentSideConnection(
            adapter,
            writer_agent,
            reader_agent,
            listening=False,
            use_unstable_protocol=True,
        )

        async def _ws_to_pipe() -> None:
            try:
                async for message in websocket.iter_text():
                    writer_bridge.write(message.encode() + b"\n")
                    await writer_bridge.drain()
            finally:
                writer_bridge.close()

        async def _pipe_to_ws() -> None:
            try:
                while True:
                    line = await reader_bridge.readline()
                    if not line:
                        break
                    await websocket.send_text(line.decode().rstrip("\n"))
            except Exception:
                pass

        try:
            await asyncio.gather(
                conn.listen(),
                _ws_to_pipe(),
                _pipe_to_ws(),
                return_exceptions=True,
            )
        finally:
            await conn.close()
            writer_agent.close()
            writer_bridge.close()
            sock_agent.close()
            sock_bridge.close()

    return WebSocketRoute("/acp/ws", _handle_ws)