What is ACP ? – DEV Community – Paradise Solution Newspaper

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.

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

Source link