ACP — Agent Communication Protocol — AI Tutorials

What Is ACP?

Agent Communication Protocol (ACP) is an open standard developed by IBM as part of the BeeAI framework. Like A2A, it addresses agent interoperability — but with a different design philosophy: absolute simplicity.

Where A2A introduces Agent Cards, task lifecycles, and streaming as first-class concepts, ACP keeps the surface area minimal: an agent is just an HTTP endpoint that accepts a message and returns a message. Any HTTP client can talk to any ACP server, with no SDK required.

ACP Design Principles

REST-native — built entirely on standard HTTP semantics, no new transport layer
Multipart messages — rich, multimodal content (text, images, files, JSON) in one message
Synchronous and async — supports both blocking responses and long-running async runs
Framework-agnostic — works with LangChain, CrewAI, LangGraph, or raw Anthropic API
Minimal spec — the core protocol fits in a single page; easy to implement from scratch

ACP Core Concepts

Concept	Description
Agent	An HTTP service registered in an ACP server registry
Run	An execution of an agent with a given input — the async equivalent of a function call
Message	Input or output of a run — a list of typed Parts
Part	A unit of content: text, image, audio, video, or custom data
Run State	`created → in-progress → completed
Agent Registry	A discovery service listing available agents and their endpoints

ACP Message Structurepython

# ACP messages are multipart — each part has a type
# This allows rich, multimodal content in a single message

# Text-only message (most common)
text_message = {
    "parts": [
        {
            "content_type": "text/plain",
            "content": "Summarize the key findings from this quarter's sales data"
        }
    ]
}

# Multimodal message (text + image + JSON data)
multimodal_message = {
    "parts": [
        {
            "content_type": "text/plain",
            "content": "Analyze this chart and explain the trend"
        },
        {
            "content_type": "image/png",
            "content": "<base64-encoded-image-data>"
        },
        {
            "content_type": "application/json",
            "content": {
                "chart_type": "line",
                "period": "Q1-Q4 2024",
                "metric": "revenue"
            }
        }
    ]
}

# Agent output — also multipart
agent_output = {
    "parts": [
        {
            "content_type": "text/plain",
            "content": "The chart shows a steady upward trend in revenue..."
        },
        {
            "content_type": "application/json",
            "content": {
                "trend": "upward",
                "growth_rate": "12.3%",
                "peak_quarter": "Q4"
            }
        }
    ]
}

Building an ACP Agent Serverpython

# acp_server.py — ACP-compliant agent server using FastAPI
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from typing import Optional, Literal
import uuid, asyncio
from anthropic import Anthropic

app = FastAPI(title="ACP Research Agent")
anthropic_client = Anthropic()

# -------------------------------------------------------
# ACP Data Models
# -------------------------------------------------------
class MessagePart(BaseModel):
    content_type: str
    content: str | dict | list

class Message(BaseModel):
    parts: list[MessagePart]

class RunCreateRequest(BaseModel):
    agent_id: str
    input: Message

class RunStatus(BaseModel):
    run_id: str
    agent_id: str
    status: Literal["created", "in-progress", "completed", "failed"]
    input: Message
    output: Optional[Message] = None
    error: Optional[str] = None

# In-memory run store (use Redis/DB in production)
runs: dict[str, RunStatus] = {}

# -------------------------------------------------------
# Agent Registry — lists available agents
# -------------------------------------------------------
AGENT_REGISTRY = {
    "research-agent": {
        "id": "research-agent",
        "name": "Research Agent",
        "description": "Researches topics and writes comprehensive reports",
        "version": "1.0.0",
        "input_content_types": ["text/plain"],
        "output_content_types": ["text/plain", "application/json"],
    },
    "summarizer-agent": {
        "id": "summarizer-agent",
        "name": "Summarizer Agent",
        "description": "Condenses long text into key bullet points",
        "version": "1.0.0",
        "input_content_types": ["text/plain"],
        "output_content_types": ["text/plain"],
    },
}

@app.get("/agents")
async def list_agents():
    """ACP agent discovery endpoint."""
    return {"agents": list(AGENT_REGISTRY.values())}

@app.get("/agents/{agent_id}")
async def get_agent(agent_id: str):
    if agent_id not in AGENT_REGISTRY:
        raise HTTPException(status_code=404, detail="Agent not found")
    return AGENT_REGISTRY[agent_id]

# -------------------------------------------------------
# Run lifecycle endpoints
# -------------------------------------------------------
@app.post("/runs", response_model=RunStatus)
async def create_run(request: RunCreateRequest, background_tasks: BackgroundTasks):
    """Create a new agent run (async execution)."""
    if request.agent_id not in AGENT_REGISTRY:
        raise HTTPException(status_code=404, detail="Agent not found")

    run_id = str(uuid.uuid4())
    run = RunStatus(
        run_id=run_id,
        agent_id=request.agent_id,
        status="created",
        input=request.input,
    )
    runs[run_id] = run

    # Execute asynchronously in background
    background_tasks.add_task(execute_run, run_id, request.agent_id, request.input)
    return run

@app.get("/runs/{run_id}", response_model=RunStatus)
async def get_run(run_id: str):
    """Poll run status and retrieve output when complete."""
    if run_id not in runs:
        raise HTTPException(status_code=404, detail="Run not found")
    return runs[run_id]

async def execute_run(run_id: str, agent_id: str, input_message: Message):
    """The actual agent execution logic."""
    runs[run_id].status = "in-progress"
    try:
        # Extract text from input parts
        text_parts = [p.content for p in input_message.parts
                      if p.content_type == "text/plain" and isinstance(p.content, str)]
        user_text = " ".join(text_parts)

        # Call the LLM
        response = anthropic_client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system="You are a helpful AI assistant. Provide thorough, well-structured responses.",
            messages=[{"role": "user", "content": user_text}],
        )

        result_text = response.content[0].text

        # Build ACP output message
        runs[run_id].output = Message(parts=[
            MessagePart(content_type="text/plain", content=result_text)
        ])
        runs[run_id].status = "completed"

    except Exception as e:
        runs[run_id].status = "failed"
        runs[run_id].error = str(e)

# Synchronous (blocking) endpoint — for simple request-response
@app.post("/runs/sync")
async def create_run_sync(request: RunCreateRequest):
    """Create and immediately await a run (synchronous)."""
    run_id = str(uuid.uuid4())
    run = RunStatus(
        run_id=run_id,
        agent_id=request.agent_id,
        status="created",
        input=request.input,
    )
    runs[run_id] = run
    await execute_run(run_id, request.agent_id, request.input)
    return runs[run_id]

# Run: uvicorn acp_server:app --port 8001 --reload

ACP Client — Discover, Run, and Retrievepython

# acp_client.py
import httpx, asyncio, time

ACP_SERVER = "http://localhost:8001"

async def list_agents() -> list[dict]:
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"{ACP_SERVER}/agents")
        agents = resp.json()["agents"]
        for agent in agents:
            print(f"- {agent['id']}: {agent['description']}")
        return agents

async def run_agent_sync(agent_id: str, text: str) -> str:
    """Use the synchronous endpoint for quick tasks."""
    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(
            f"{ACP_SERVER}/runs/sync",
            json={
                "agent_id": agent_id,
                "input": {
                    "parts": [{"content_type": "text/plain", "content": text}]
                }
            }
        )
        run = resp.json()
        if run["status"] == "completed":
            return run["output"]["parts"][0]["content"]
        raise RuntimeError(f"Run failed: {run.get('error')}")

async def run_agent_async(agent_id: str, text: str) -> str:
    """Use the async endpoint with polling for long-running tasks."""
    async with httpx.AsyncClient(timeout=10) as client:
        # Create the run
        resp = await client.post(
            f"{ACP_SERVER}/runs",
            json={
                "agent_id": agent_id,
                "input": {"parts": [{"content_type": "text/plain", "content": text}]}
            }
        )
        run_id = resp.json()["run_id"]
        print(f"Run created: {run_id}")

        # Poll until done
        for _ in range(30):
            await asyncio.sleep(2)
            poll_resp = await client.get(f"{ACP_SERVER}/runs/{run_id}")
            run = poll_resp.json()
            print(f"  Status: {run['status']}")
            if run["status"] == "completed":
                return run["output"]["parts"][0]["content"]
            if run["status"] == "failed":
                raise RuntimeError(run.get("error", "Unknown error"))

        raise TimeoutError("Run timed out")

async def main():
    print("=== Available Agents ===")
    await list_agents()

    print("\n=== Synchronous Run ===")
    result = await run_agent_sync(
        "research-agent",
        "Explain the Agent Communication Protocol (ACP) in 3 bullet points"
    )
    print(result)

    print("\n=== Asynchronous Run ===")
    result = await run_agent_async(
        "research-agent",
        "Write a detailed comparison of ACP vs A2A protocol for AI agents"
    )
    print(result)

asyncio.run(main())

ACP vs A2A: Choosing the Right Protocol

Dimension	ACP	A2A
Complexity	Minimal — just HTTP + JSON	Richer — Agent Cards, SSE, push notifications
Discovery	Agent Registry (centralized)	Agent Cards at /.well-known/ (distributed)
Streaming	Not in core spec	First-class SSE streaming
Multimodal	First-class multipart messages	Text-focused, extensible
Ecosystem	IBM BeeAI ecosystem	Google-backed, growing
Best for	Internal microservice-style agents	Cross-org, internet-scale agent networks
Learning curve	Very low	Moderate

Decision Guide

Use ACP when:

You want the simplest possible implementation
Agents are within your organization or trusted network
You need strong multimodal support (images, audio, files)
You're already using the BeeAI framework

Use A2A when:

Agents cross organizational boundaries
You need real-time streaming progress
Agents need to advertise capabilities publicly
You want distributed discovery without a central registry

BeeAI Framework Integration

ACP is the native protocol of IBM's BeeAI framework. If you're using BeeAI, ACP is built in:

from beeai_framework.agents.react import ReActAgent
from beeai_framework.backend.anthropic import AnthropicChatModel
from beeai_framework.tools.search import DuckDuckGoSearchTool
from beeai_framework.serve.acp import ACPServer

# Create a BeeAI agent
agent = ReActAgent(
    llm=AnthropicChatModel("claude-sonnet-4-6"),
    tools=[DuckDuckGoSearchTool()],
)

# Serve it via ACP — one line
server = ACPServer(agents={"research": agent})
server.run(port=8001)

Knowledge check

What distinguishes ACP's message model from other protocols?

Summary

ACP takes a "keep it simple" philosophy to agent interoperability:

REST-native — works with any HTTP client, no SDK required
Multipart messages — rich multimodal content as first-class citizens
Sync and async runs — flexible execution model for any latency requirement
Centralized registry — easy discovery without distributed agent card hosting
BeeAI native — tight integration with IBM's open-source agent framework

In the next chapter, we'll explore ANP (Agent Network Protocol) — a decentralized approach using DIDs and JSON-LD to enable agents to find and trust each other without any central registry.

← Agentic AI