Compose Agents - Agent Communication Protocol

ACP is designed to be agnostic regarding the internal implementation details of agents. It provides a standardized interface that facilitates communication between agents, enabling seamless composition. Rather than prescribing specific frameworks, ACP emphasizes patterns over frameworks, echoing sentiments expressed in Anthropic’s insightful article on building effective agents. Central to ACP’s composability are its message structure and agent execution model. A consistent message format and the capability to invoke agents remotely are crucial for effective composition.

When to Use Composition Patterns

Prompt Chaining: When you need sequential processing where each step builds on the previous output (writing → editing → translation).
Routing: When different request types need specialized handling (customer support routing to technical/billing/general agents).
Parallelization: When independent tasks can be processed simultaneously for faster results (generating multiple translations or analyses).
Hierarchical: When you need coordination between high-level planning and specialized execution agents.

Let’s explore the implementation of these patterns using ACP.

Prompt Chaining Example

See the complete source code on GitHub.

Using ACP, prompt chaining can be implemented easily by sequentially running multiple agents and combining their outputs. The following example demonstrates chaining two agents sequentially: first, an agent generates a punchy headline for a product; next, another agent translates the headline into Spanish. Finally, the composition agent combines these results and returns them to the client.

agent.py

from collections.abc import AsyncGenerator

import beeai_framework
from acp_sdk import Message
from acp_sdk.client import Client
from acp_sdk.models import MessagePart
from acp_sdk.server import Context, Server
from beeai_framework.backend.chat import ChatModel
from beeai_framework.agents.react import ReActAgent
from beeai_framework.memory import TokenMemory

server = Server()

async def run_agent(agent: str, input: str) -> list[Message]:
    async with Client(base_url="http://localhost:8000") as client:
        run = await client.run_sync(
            agent=agent,
            input=[Message(parts=[MessagePart(content=input, content_type="text/plain")])]
        )

    return run.output

@server.agent(name="translation")
async def translation_agent(input: list[Message]) -> AsyncGenerator:
    llm = ChatModel.from_name("ollama:llama3.1:8b")

    agent = ReActAgent(llm=llm, tools=[], memory=TokenMemory(llm))
    response = await agent.run(prompt="Translate the given text to Spanish. The text is: " + str(input))

    yield MessagePart(content=response.result.text)

@server.agent(name="marketing_copy")
async def marketing_copy_agent(input: list[Message]) -> AsyncGenerator:
    llm = ChatModel.from_name("ollama:llama3.1:8b")

    agent = ReActAgent(llm=llm, tools=[], memory=TokenMemory(llm))
    response = await agent.run(prompt="You are able to generate punchy headlines for a marketing campaign. Provide punchy headline to sell the specified product on users eshop. The product is: " + str(input))

    yield MessagePart(content=response.result.text)


@server.agent(name="assistant")
async def main_agent(input: list[Message], context: Context) -> AsyncGenerator:
    marketing_copy = await run_agent("marketing_copy", str(input))
    translated_marketing_copy = await run_agent("translation", str(marketing_copy))

    yield MessagePart(content=str(marketing_copy[0]))
    yield MessagePart(content=str(translated_marketing_copy[0]))

server.run()

Key points:

While the example uses a single ACP server to expose multiple agents for simplicity, practical implementations may involve distributed architectures.
The run_agent function enables remote invocation of agents through ACP.
While the current example demonstrates agents that only accept and produce text, practical implementations may include more sophisticated use cases involving various types of artifacts.

Intelligent Routing Example

See the complete source code on GitHub.

Routing enables dynamic agent selection based on request content. A router agent analyzes incoming requests and forwards them to the most appropriate specialist agent. The following example exposes ACP agents as tools to a router agent. The router agent evaluates the original request and forwards it to the appropriate agent based on its assessment.

from collections.abc import AsyncGenerator

from acp_sdk import Message
from acp_sdk.models import MessagePart
from acp_sdk.server import Context, Server
from beeai_framework.backend.chat import ChatModel
from beeai_framework.agents.react import ReActAgent
from beeai_framework.memory import TokenMemory
from beeai_framework.utils.dicts import exclude_none
from translation_tool import TranslationTool

server = Server()

@server.agent(name="translation_spanish")
async def translation_spanish_agent(input: list[Message]) -> AsyncGenerator:
llm = ChatModel.from_name("ollama:llama3.1:8b")
print("Translation Spanish agent")

    agent = ReActAgent(llm=llm, tools=[], memory=TokenMemory(llm))
    response = await agent.run(prompt="Translate the given text to Spanish. The text is: " + str(input))

    yield MessagePart(content=response.result.text)

@server.agent(name="translation_french")
async def translation_french_agent(input: list[Message]) -> AsyncGenerator:
llm = ChatModel.from_name("ollama:llama3.1:8b")

    agent = ReActAgent(llm=llm, tools=[], memory=TokenMemory(llm))
    response = await agent.run(prompt="Translate the given text to French. The text is: " + str(input))

    yield MessagePart(content=response.result.text)

@server.agent(name="router")
async def main_agent(input: list[Message], context: Context) -> AsyncGenerator:
llm = ChatModel.from_name("ollama:llama3.1:8b")

    agent = ReActAgent(
        llm=llm,
        tools=[TranslationTool()],
        templates={
            "system": lambda template: template.update(
                defaults=exclude_none({
                    "instructions": """
                        Translate the given text to either Spanish or French using the translation tool.
                        Return only the result from the tool as it is, don't change it.
                    """,
                    "role": "system"
                })
            )
        },
        memory=TokenMemory(llm)
    )

    prompt = (str(input[0]))
    response = await agent.run(prompt)

    yield MessagePart(content=response.result.text)

server.run()

from pydantic import BaseModel, Field
from enum import Enum

from acp_sdk import Message
from acp_sdk.client import Client
from acp_sdk.models import MessagePart
from beeai_framework.tools.tool import Tool
from beeai_framework.tools.types import ToolRunOptions
from beeai_framework.context import RunContext
from beeai_framework.emitter import Emitter
from beeai_framework.tools import ToolOutput
from beeai_framework.utils.strings import to_json

async def run_agent(agent: str, input: str) -> list[Message]:
    async with Client(base_url="http://localhost:8000") as client:
        run = await client.run_sync(
            agent=agent, input=[Message(parts=[MessagePart(content=input, content_type="text/plain")])]
        )

    return run.output

class Language(str, Enum):
    spanish = 'spanish'
    french = 'french'

class TranslateToolInput(BaseModel):
    text: str = Field(description="The text to translate")
    language: Language = Field(description="The language to translate the text to")

class TranslateToolResult(BaseModel):
    text: str = Field(description="The translated text")

class TranslateToolOutput(ToolOutput):
    result: TranslateToolResult = Field(description="Translation result")

    def get_text_content(self) -> str:
        return to_json(self.result)

    def is_empty(self) -> bool:
        return self.result.text == ""


    def __init__(self, result: TranslateToolResult) -> None:
        super().__init__()
        self.result = result


class TranslationTool(Tool[TranslateToolInput, ToolRunOptions, TranslateToolOutput]):
    name = "Translation"
    description = "Translate the given text to the specified language"
    input_schema = TranslateToolInput

    def _create_emitter(self) -> Emitter:
        return Emitter.root().child(
            namespace=["tool", "translate"],
            creator=self,
        )


    async def _run(self, input: TranslateToolInput, options: ToolRunOptions | None, context: RunContext) -> TranslateToolOutput:
        if input.language == Language.spanish:
            result = await run_agent("translation_spanish", input.text)
        elif input.language == Language.french:
            result = await run_agent("translation_french", input.text)

        return TranslateToolOutput(result=TranslateToolResult(text=str(result[0])))

curl -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json" \
  -d '{
        "agent_name": "router",
        "input": [
          {
            "role": "user",
            "parts": [
              {
                "content_type": "text/plain",
                "content": "Translate text \"Hello world\" to Spanish."
              }
            ]
          }
        ]
      }'

Key points:

The run_agent function enables remote invocation of agents through ACP.
The router agent is provided with a TranslationTool, which can invoke both translation_french and translation_spanish agents via ACP by using run_agent.
Based on the user’s input, the router decides which agent to invoke to fulfill the user’s request.

Parallelization Example

See the complete source code on GitHub.

Parallelization executes multiple agents simultaneously to reduce overall processing time. Use this pattern when you need the same input processed by different agents or when tasks are independent. This example uses asyncio.gather to execute multiple remote agent calls concurrently via ACP. The process then awaits both responses, effectively blocking until all agents return their results.

from collections.abc import AsyncGenerator

from acp_sdk import Message
from acp_sdk.client.client import Client
from acp_sdk.models import MessagePart
from acp_sdk.server import Context, Server
from beeai_framework.agents.react import ReActAgent
from beeai_framework.backend.chat import ChatModel
from beeai_framework.memory import TokenMemory

import asyncio

server = Server()


async def run_agent(agent: str, input: str) -> list[Message]:
    async with Client(base_url="http://localhost:8000") as client:
        run = await client.run_sync(
            agent=agent, input=input
        )

    return run.output


@server.agent()
async def translation_spanish(input: list[Message]) -> AsyncGenerator:
    llm = ChatModel.from_name("ollama:llama3.1:8b")

    agent = ReActAgent(llm=llm, tools=[], memory=TokenMemory(llm))
    response = await agent.run(
        prompt="Translate the given English text to Spanish. Return only the translated text. The text is: "
        + str(input)
    )

    yield MessagePart(content=response.result.text)


@server.agent()
async def translation_french(input: list[Message]) -> AsyncGenerator:
    llm = ChatModel.from_name("ollama:llama3.1:8b")

    agent = ReActAgent(llm=llm, tools=[], memory=TokenMemory(llm))
    response = await agent.run(
        prompt="Translate the given English text to French. Return only the translated text. The text is: " + str(input)
    )

    yield MessagePart(content=response.result.text)


@server.agent()
async def aggregator(input: list[Message], context: Context) -> AsyncGenerator:
    spanish_result, english_result = await asyncio.gather(
        run_agent("translation_spanish", str(input[0])), run_agent("translation_french", str(input[0]))
    )

    yield MessagePart(content=str(spanish_result[0]), language="Spanish")
    yield MessagePart(content=str(english_result[0]), language="French")


server.run()

curl -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json" \
  -d '{
        "agent_name": "aggregator",
        "input": [
          {
            "role": "user",
            "parts": [
              {
                "content_type": "text/plain",
                "content": "Translate text \"Hello world\"."
              }
            ]
          }
        ]
      }'

Key points:

The run_agent function enables remote invocation of agents through ACP.
The aggregator agent invokes both translation_french and translation_spanish in parallel using asyncio.gather

​When to Use Composition Patterns

​Prompt Chaining Example

​Intelligent Routing Example

​Parallelization Example

When to Use Composition Patterns

Prompt Chaining Example

Intelligent Routing Example

Parallelization Example