Server

ACPAgent gives your agent a user interface you do not have to write.

Wrap an existing Agent, serve it over the Agent Client Protocol (ACP), and it becomes drivable from any ACP client — editors like Zed, JetBrains, VS Code and Neovim, note-taking apps, notebook kernels, chat bridges. Streamed output, tool calls shown as they happen, and a working cancel button all come from the client. You write the agent; somebody else already wrote the frontend.

That is the trade this page is about: one line of code instead of a websocket server, a streaming protocol and a chat UI.

This is the mirror of the Client page, which covers AG2 driving an external CLI coding agent.

pip install "ag2[acp]"

Two roles, and who starts whom#

ACP has exactly two sides: Client and Agent. They map onto your world cleanly — the Agent is your AG2 agent, the Client is whatever wants to talk to it.

The part that surprises most people is the direction: the Client launches the Agent. You are not standing a service up and waiting for traffic. ACP's stable transport is stdio, so the Client starts your process and the two talk over the pipe between them. No ports, no URLs, nothing to deploy.

Minimal Server#

The whole integration is ACPAgent(agent).run_stdio().

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import asyncio

from ag2 import Agent
from ag2.acp import ACPAgent
from ag2.config import AnthropicConfig
from ag2.tools import tool

@tool(description="Add two integers.")
async def calc_add(a: int, b: int) -> str:
    return str(a + b)

agent = Agent(
    name="workie",
    prompt="You are a concise assistant. Use tools when they help.",
    config=AnthropicConfig(model="claude-sonnet-4-6", streaming=True),
    tools=[calc_add],
)

async def main() -> None:
    await ACPAgent(agent).run_stdio()

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

Point an ACP client at that file and it runs the agent as a subprocess:

stop_reason: end_turn
answer: 2 + 2 = 4

The same file, ready to run, is at examples/acp/server_stdio.py.

Pointing a Client at It#

Every client configures a custom agent the same way — a command, its arguments, and an environment. Zed's settings.json, for example:

{
  "agent_servers": {
    "workie": {
      "type": "custom",
      "command": "/abs/path/to/.venv/bin/python",
      "args": ["/abs/path/to/examples/acp/server_stdio.py"],
      "env": { "ANTHROPIC_API_KEY": "sk-ant-..." }
    }
  }
}

What Happens in One Turn#

A prompt is one AG2 turn through the same path Agent.ask() uses, so dynamic prompts, middleware, observers and tool events all behave exactly as they do off-protocol.

Sessions#

Each session/new gets its own conversation. Two things are isolated per session:

• History. A session never sees another session's messages.
• Context variables. Seeded by value from the agent's defaults, so writes in one session are invisible to the others, and survive to that session's next prompt.

Concurrent prompts on one session queue rather than interleave, matching how MCPServer already treats overlapping calls on one session.

Cancelling#

session/cancel names a session, so that is exactly how far it reaches: the turn running there stops, anything queued behind it is dropped, and every other session carries on untouched.

Two details worth knowing:

• The prompt still returns — with stop_reason="cancelled", not an error.
• Whatever already reached the client stays. Stopping the work does not un-say what was already said.

Cancelling four seconds into a long streamed answer:

stop_reason: cancelled
updates kept: 12

How many updates survive depends on how much the model had streamed by then. The point is that it is not zero — on a streaming=True config. On the default it is zero, for the reason above: nothing had been sent yet.

If the cancellation lands between a tool call and its result, the transcript is repaired before the session is released, so the next prompt still sends a valid conversation to the model.

Configuring the Agent#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from ag2 import Agent
from ag2.acp import ACPAgent, PromptContent, SessionConfig, StaticTokenAuth
from ag2.config import AnthropicConfig

agent = Agent(name="workie", config=AnthropicConfig(model="claude-haiku-4-5-20251001"))

acp_agent = ACPAgent(
    agent,
    name="workie",
    version="1.0.0",
    sessions=SessionConfig(
        max_sessions=256,
        ttl=3600,
        max_queued=4,
        max_concurrent_turns=4,
        max_active_prompts=32,
    ),
    auth=StaticTokenAuth("shared-secret"),
    prompt_content=PromptContent(image=True, audio=False, embedded_context=True),
    stream_thoughts=False,
)

Bounds#

SessionConfig bounds both a single session and the connection as a whole.

The last two are what keep one client from opening every session and starting a paid turn in each. max_concurrent_turns bounds spend; max_active_prompts bounds how many request handlers can be parked at once.

Multimodal content#

PromptContent declares what the model behind your agent can actually take.

Whether an image or an audio clip works depends on the provider, and AG2 has no registry of provider modalities to consult — the ACP mapper will happily build an AudioInput that the Anthropic mapper then rejects. So this is a declaration by whoever deploys the agent, not a guess. Audio is off by default because most providers do not accept it.

Authentication#

Local stdio needs none: the client already launched the process. A provider becomes necessary for a remote transport, or for the ACP Registry.

1
2
3
4
5
6

from ag2 import Agent
from ag2.acp import ACPAgent, StaticTokenAuth
from ag2.config import AnthropicConfig

agent = Agent(name="workie", config=AnthropicConfig(model="claude-haiku-4-5-20251001"))
acp_agent = ACPAgent(agent, auth=StaticTokenAuth("shared-secret"))

With a provider attached, session methods are refused until the client authenticates:

auth methods: ['token']
unauthenticated session refused: ok
session opened after auth: True

Implement AuthProvider to plug in your own identity system.

What It Refuses to Pretend#

During initialize the agent declares what it supports, and the client builds its interface from that answer. Claiming a feature that is not wired is worse than admitting its absence — the client offers users a button that cannot work.

Reported as not supported: session/load, session/resume, session/fork, session/list, session/delete, and connecting to client-provided MCP servers over any transport. The client's own filesystem and terminal methods are never called either — those the client offers, so there is nothing to declare.

These are optional parts of the protocol, not missing pieces of this one. ACP grew up around coding agents, and most of that list exists to serve them — a client offering its open buffers and its terminal to an agent whose job is editing your repository.

A cwd, an additional directory or an MCP server named by a client is recorded as context and acted on by nothing. A path a client names is not authority to reach it; deciding what to honour belongs to the application embedding the agent.

Testing#

ag2.acp.testing.connect wires a real ACP client to your ACPAgent inside the test process — genuine protocol, no subprocess to manage and no port to bind. Pair it with TestConfig so no provider is called.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import acp
import pytest
from acp import schema

from ag2 import Agent
from ag2.acp import ACPAgent
from ag2.acp.testing import connect
from ag2.testing import TestConfig

@pytest.mark.asyncio
async def test_the_agent_answers() -> None:
    agent = Agent(name="workie", config=TestConfig("200"))

    async with connect(ACPAgent(agent)) as (client, recorder):
        session = await client.new_session(cwd=".")
        response = await client.prompt(
            session_id=session.session_id,
            prompt=[acp.text_block("what's 100 + 100")],
        )

    assert response.stop_reason == "end_turn"
    texts = [
        update.content.text
        for update in recorder.updates_for(session.session_id)
        if isinstance(update, schema.AgentMessageChunk)
    ]
    assert texts == ["200"]

recorder is a RecordingClient: it captures every session/update the agent pushed, so you can assert on what the client actually saw and in what order.