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.

The protocol emphasizes the importance of patterns over frameworks, echoing sentiments expressed in Anthropic’s insightful article.

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.

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

​
Example: Prompt Chaining

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.

​
Example: Routing

See the complete source code on GitHub.

Routing is a concept where a router (often an LLM) determines which agent should handle a particular request.

The following example illustrates routing by exposing 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()

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.

​
Example: Paralelization

See the complete source code on GitHub.

Parallelization is conceptually similar to prompt chaining, with one key difference: instead of invoking agents sequentially, they are called in parallel.

In practice, this means using 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()

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