Skip to content

Latest commit

 

History

History
99 lines (73 loc) · 3.25 KB

File metadata and controls

99 lines (73 loc) · 3.25 KB

Agent Client Protocol (Python)

A Python implementation of the Agent Client Protocol (ACP). Use it to build agents that communicate with ACP-capable clients (e.g. Zed) over stdio.

This library version is compatible with the corresponding ACP version. However, we lack sufficient resources to release each version. Welcome to contribute!

Install

pip install agent-client-protocol
# or
uv add agent-client-protocol

Development (contributors)

make install   # set up venv
ACP_SCHEMA_VERSION=<ref> make gen-all   # generate meta.py & schema.py
make check     # lint + typecheck
make test      # run tests

Minimal agent example

See a complete streaming echo example in examples/echo_agent.py. It streams back each text block using session/update and ends the turn.

import asyncio

from acp import (
    Agent,
    AgentSideConnection,
    InitializeRequest,
    InitializeResponse,
    NewSessionRequest,
    NewSessionResponse,
    PromptRequest,
    PromptResponse,
    SessionNotification,
    stdio_streams,
)
from acp.schema import TextContentBlock, AgentMessageChunk


class EchoAgent(Agent):
    def __init__(self, conn):
        self._conn = conn

    async def initialize(self, params: InitializeRequest) -> InitializeResponse:
        return InitializeResponse(protocolVersion=params.protocolVersion)

    async def newSession(self, params: NewSessionRequest) -> NewSessionResponse:
        return NewSessionResponse(sessionId="sess-1")

    async def prompt(self, params: PromptRequest) -> PromptResponse:
        for block in params.prompt:
            text = block.get("text", "") if isinstance(block, dict) else getattr(block, "text", "")
            await self._conn.sessionUpdate(
                SessionNotification(
                    sessionId=params.sessionId,
                    update=AgentMessageChunk(
                        sessionUpdate="agent_message_chunk",
                        content=TextContentBlock(type="text", text=text),
                    ),
                )
            )
        return PromptResponse(stopReason="end_turn")


async def main() -> None:
    reader, writer = await stdio_streams()
    AgentSideConnection(lambda conn: EchoAgent(conn), writer, reader)
    await asyncio.Event().wait()


if __name__ == "__main__":
    asyncio.run(main())

Run this executable from your ACP-capable client (e.g. configure Zed to launch it). The library takes care of the stdio JSON-RPC transport.

Example: Mini SWE Agent bridge

A minimal ACP bridge for mini-swe-agent is provided under examples/mini_swe_agent. It demonstrates:

  • Parsing a prompt from ACP content blocks
  • Streaming agent output via session/update
  • Mapping command execution to tool_call and tool_call_update

Documentation