What is ACP (Agent Communication Protocol)?
The Agent Communication Protocol (ACP) is an open standard designed to enable interoperable, low-latency communication between AI agents, regardless of the framework, programming language, or runtime environment they’re built on
Think of ACP as the HTTP for AI agents a lightweight, REST-native protocol that allows agents to discover, message, and coordinate with each other in real time. It abstracts away framework-specific details and provides a unified interface for agent interaction, making it ideal for building modular, composable, and scalable multi-agent systems.
Why ACP Matters
Modern AI systems often rely on specialized agents for tasks like retrieval, reasoning, classification, and tool use. However, these agents are typically siloed within their own ecosystems (LangChain, CrewAI, BeeAI, etc.), making integration difficult.
ACP solves this by:
- Standardizing communication across agents
- Supporting multimodal messaging (text, images, embeddings, etc.)
- Enabling local-first orchestration without cloud dependencies
- Facilitating agent discovery and collaboration across organizations
ACP Architecture
ACP supports multiple deployment patterns :
1. Single-Agent Server
- A client communicates with one agent via a REST interface.
- Ideal for simple setups and debugging.
2. Multi-Agent Server
- One server hosts multiple agents.
- Each agent is addressable via metadata-based routing.
3. Distributed Multi-Server
- Multiple servers host agents independently.
- Enables scalability, fault isolation, and flexible deployment.
4. Router Agent Pattern
- A central agent decomposes tasks and delegates to specialized agents.
- Supports orchestration, parallelism, and aggregation.
ACP Endpoints
ACP defines a minimal set of RESTful endpoints for agent interaction. Here’s a breakdown:
Run Lifecycle Endpoints
POST /runs Starts a new agent run. Requires agent_name, input, and optional mode (sync, async, stream).
GET /runs/{run_id} Retrieves the current state and output of a run.
POST /runs/{run_id} Resumes a paused run (in awaiting state).
POST /runs/{run_id}/cancel Requests cancellation of an ongoing run.
Agent Metadata & Discovery
GET /agents Lists all available agents.
GET /agents/{agent_id} Retrieves metadata about a specific agent.
Message Exchange
POST /agents/{agent_id}/messages Sends a message to an agent.
GET /agents/{agent_id}/messages/{message_id}/response Retrieves the response to a message.
ACP Server default Endpoints
Key Features of ACP
- Framework Agnostic: Works with LangChain, CrewAI, AutoGen, or custom implementations
- REST-Based: Standard HTTP communication protocol
- Async Support: Both synchronous and asynchronous transmission
- State Management: Supports stateless and stateful operation patterns
- Streaming: Real-time interaction streaming capabilities
- Discovery: Online and offline agent discovery mechanisms
- Open Governance: Community-driven development and standards
ACP Server
"""
ACP Server - SreeniParrotAgent Implementation
This server implements a simple ACP (Agent Communication Protocol) server with a single agent
that echoes back received messages after a 30-second processing delay.
Dependencies:
- acp_sdk: The ACP SDK for building agent servers
- asyncio: For asynchronous operations
Usage:
python main.py
The server will start on http://0.0.0.0:8000
"""
import asyncio
from acp_sdk.server import Server
from typing import AsyncGenerator
from acp_sdk import MessagePart
# Initialize the ACP server instance
server = Server()
@server.agent(
name="SreeniParrotAgent", # Unique identifier for the agent
description="An agent that echoes back the received message.", # Human-readable description
metadata={
"version": "1.0", # Agent version
"author": {
"name": "Sreeni", # Author name
"email": "sreeni@example.com" # Author contact
}
}
)
async def SreeniParrotAgent(messages: list) -> AsyncGenerator[MessagePart, None]:
"""
Main agent function that processes incoming messages.
Args:
messages (list): List of message objects containing input data
Yields:
MessagePart: Processed output messages
Processing Flow:
1. Receives messages from client
2. Waits 30 seconds (simulating processing time)
3. Echoes back the original content
"""
for message in messages:
# Process each message in the input
for part in message.parts:
# Simulate processing delay (30 seconds)
await asyncio.sleep(30)
# Echo back the received content as output
yield MessagePart(
name="output", # Output identifier
content_type="text", # Content type (text in this case)
content=part.content # The actual content to echo back
)
if __name__ == "__main__":
# Start the server when script is run directly
server.run(host="0.0.0.0", port=8000)
ACP client with python client SDK
"""
ACP Client - Endpoint Testing Suite
This client tests all available endpoints of the ACP server and displays
comprehensive status information and response content.
Dependencies:
- requests: HTTP library for making API calls
Usage:
python client.py
Prerequisites:
- ACP server must be running on http://0.0.0.0:8000
"""
import requests
import json
# Base URL for the ACP server
BASE_URL = "http://0.0.0.0:8000"
# ANSI color codes for terminal output
class Colors:
BLUE = '\033[94m' # Blue for URLs
GREEN = '\033[92m' # Green for success
YELLOW = '\033[93m' # Yellow for warnings
RED = '\033[91m' # Red for errors
PURPLE = '\033[95m' # Purple for info
CYAN = '\033[96m' # Cyan for status
BOLD = '\033[1m' # Bold text
UNDERLINE = '\033[4m' # Underlined text
END = '\033[0m' # Reset color
# Example payload for creating a new run
# This payload tells the server to process "Hello, World!" through SreeniParrotAgent
example_payload = {
"agent_name": "SreeniParrotAgent", # Which agent to use
"input": [
{
"parts": [
{
"name": "input", # Input identifier
"content_type": "text", # Type of content
"content": "Hello, World!" # Actual message content
}
]
}
]
}
# Payload for resuming an existing run
# This allows adding more messages to an ongoing conversation
resume_payload = {
"await_resume": {
"enabled": True, # Enable resume functionality
"message": {
"parts": [
{
"name": "input",
"content_type": "text",
"content": "Resuming run"
}
]
}
},
"mode": "sync", # Synchronous processing mode
"messages": [
{
"parts": [
{
"name": "input",
"content_type": "text",
"content": "Hello again!" # Additional message content
}
]
}
]
}
def print_detailed_status(data, endpoint_name):
"""
Display comprehensive status information from API responses.
Args:
data: Response data from the API
endpoint_name (str): Name of the endpoint being tested
"""
print(f"\n{Colors.BOLD}{endpoint_name} - Detailed Status:{Colors.END}")
print("-" * 50)
if isinstance(data, dict):
for key, value in data.items():
# Handle status information specially
if key == "status_info" and isinstance(value, dict):
print(f" {Colors.CYAN}STATUS INFORMATION:{Colors.END}")
for status_key, status_value in value.items():
print(f" {status_key}: {status_value}")
# Handle agent lists specially
elif key == "agents" and isinstance(value, list):
print(f" 🤖 {Colors.PURPLE}{key.upper()}:{Colors.END}")
for agent in value:
print(f" Name: {agent.get('name')}")
print(f" Status: {agent.get('metadata', {}).get('status', 'N/A')}")
print(f" Version: {agent.get('metadata', {}).get('version', 'N/A')}")
# Handle output content specially
elif key == "output" and isinstance(value, list):
print(f" �� {Colors.GREEN}{key.upper()}:{Colors.END}")
for i, output in enumerate(value):
print(f" Output {i+1}: {output.get('content', 'N/A')}")
if 'metadata' in output and 'status_info' in output['metadata']:
print(f" Status: {output['metadata']['status_info']}")
# Display other data normally
else:
print(f" {Colors.YELLOW}{key.upper()}:{Colors.END} {value}")
else:
print(f" Content: {data}")
# ============================================================================
# ENDPOINT FUNCTIONS
# ============================================================================
def list_agents():
"""
GET /agents - Retrieve list of all available agents
"""
endpoint_url = f"{BASE_URL}/agents"
print(f"\n�� Testing: {Colors.BLUE}{Colors.BOLD}GET {endpoint_url}{Colors.END}")
response = requests.get(endpoint_url)
print(f"List Agents: {Colors.CYAN}{response.status_code}{Colors.END}")
if response.status_code == 200:
data = response.json()
print_detailed_status(data, "List Agents")
def read_agent(name):
"""
GET /agents/{name} - Get detailed information about a specific agent
Args:
name (str): Name of the agent to retrieve
"""
endpoint_url = f"{BASE_URL}/agents/{name}"
print(f"\n�� Testing: {Colors.BLUE}{Colors.BOLD}GET {endpoint_url}{Colors.END}")
response = requests.get(endpoint_url)
print(f"Read Agent {name}: {Colors.CYAN}{response.status_code}{Colors.END}")
if response.status_code == 200:
data = response.json()
print_detailed_status(data, f"Read Agent {name}")
def ping():
"""
GET /ping - Health check endpoint to verify server is running
"""
endpoint_url = f"{BASE_URL}/ping"
print(f"\n�� Testing: {Colors.BLUE}{Colors.BOLD}GET {endpoint_url}{Colors.END}")
response = requests.get(endpoint_url)
print(f"Ping: {Colors.CYAN}{response.status_code}{Colors.END}")
data = response.json()
print_detailed_status(data, "Ping")
def create_run(payload):
"""
POST /runs - Create a new run with the specified agent and input
Args:
payload (dict): Run configuration including agent name and input data
Returns:
dict: Run response data if successful, None otherwise
"""
endpoint_url = f"{BASE_URL}/runs"
print(f"\n�� Testing: {Colors.BLUE}{Colors.BOLD}POST {endpoint_url}{Colors.END}")
response = requests.post(endpoint_url, json=payload)
print(f"Create Run: {Colors.CYAN}{response.status_code}{Colors.END}")
if response.status_code == 200:
data = response.json()
print_detailed_status(data, "Create Run")
return data
return None
def read_run(run_id):
"""
GET /runs/{run_id} - Get status and output of a specific run
Args:
run_id (str): Unique identifier of the run to retrieve
"""
endpoint_url = f"{BASE_URL}/runs/{run_id}"
print(f"\n�� Testing: {Colors.BLUE}{Colors.BOLD}GET {endpoint_url}{Colors.END}")
response = requests.get(endpoint_url)
print(f"Read Run {run_id}: {Colors.CYAN}{response.status_code}{Colors.END}")
if response.status_code == 200:
data = response.json()
print_detailed_status(data, f"Read Run {run_id}")
def resume_run(run_id, payload):
"""
POST /runs/{run_id} - Resume an existing run with additional input
Args:
run_id (str): Unique identifier of the run to resume
payload (dict): Additional input data for the run
"""
endpoint_url = f"{BASE_URL}/runs/{run_id}"
print(f"\n�� Testing: {Colors.BLUE}{Colors.BOLD}POST {endpoint_url}{Colors.END}")
response = requests.post(endpoint_url, json=payload)
print(f"Resume Run {run_id}: {Colors.CYAN}{response.status_code}{Colors.END}")
if response.status_code == 200:
data = response.json()
print_detailed_status(data, f"Resume Run {run_id}")
else:
print(f"{Colors.RED}Error: {response.text}{Colors.END}")
# ============================================================================
# MAIN EXECUTION
# ============================================================================
if __name__ == "__main__":
print(f"{Colors.BOLD}Testing ACP Endpoints with Full Status Display{Colors.END}")
print("=" * 60)
# Test basic server functionality
print(f"\n{Colors.PURPLE}📡 Testing Basic Server Endpoints...{Colors.END}")
ping()
# Test agent management endpoints
print(f"\n{Colors.PURPLE}🤖 Testing Agent Endpoints...{Colors.END}")
list_agents()
read_agent("SreeniParrotAgent")
# Test run management endpoints
print(f"\n{Colors.PURPLE}🔄 Testing Run Endpoints...{Colors.END}")
# Create a new run and get the run ID for further testing
print(f"\n{Colors.PURPLE} Creating a new run to test run-specific endpoints...{Colors.END}")
run_result = create_run(example_payload)
if run_result and 'run_id' in run_result:
run_id = run_result['run_id']
# Test reading the created run
read_run(run_id)
# Test resuming the run with additional input
resume_run(run_id, resume_payload)
else:
print(f"{Colors.RED}❌ Failed to create run - cannot test run-specific endpoints{Colors.END}")
Output
Conclusion
Agent Communication Protocol (ACP) represents a significant advancement in AI agent development, addressing the critical need for interoperability and standardized communication. By providing a framework-agnostic approach to agent communication, ACP enables developers to build more robust, scalable, and collaborative AI systems.
The protocol's support for both stateful and stateless operations, combined with its REST-based architecture, makes it an ideal choice for integrating AI agents into modern microservices and event-driven systems. As the AI landscape continues to evolve, ACP's open governance and community-driven development ensure it remains relevant and adaptable to emerging needs.
For AI engineers and developers looking to build scalable, interoperable agent systems, ACP provides the foundation needed to overcome current fragmentation challenges and create more dynamic AI solutions.
Thanks
Sreeni Ramadorai
Top comments (0)