> ## Documentation Index
> Fetch the complete documentation index at: https://agentcommunicationprotocol.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Compose Agents

> Learn how to build common composition patterns

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](https://www.anthropic.com/engineering/building-effective-agents) 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

<Tip>
  See the complete source code on
  [GitHub](https://github.com/i-am-bee/acp/tree/main/examples/python/beeai-prompt-chaining).
</Tip>

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.

```python agent.py theme={null}
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](/core-concepts/message-structure#artifacts).

## Intelligent Routing Example

<Tip>
  See the complete source code on
  [GitHub](https://github.com/i-am-bee/acp/tree/main/examples/python/beeai-routing).
</Tip>

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.

<CodeGroup>
  ```python agent.py theme={null}
  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()

  ```

  ```python translation_tool.py theme={null}
  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])))
  ```

  ```bash cURL theme={null}
  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."
                }
              ]
            }
          ]
        }'
  ```
</CodeGroup>

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

<Tip>
  See the complete source code on
  [GitHub](https://github.com/i-am-bee/acp/tree/main/examples/python/beeai-parallelization).
</Tip>

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.

<CodeGroup>
  ```python agent.py theme={null}
  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()
  ```

  ```bash cURL theme={null}
  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\"."
                }
              ]
            }
          ]
        }'
  ```
</CodeGroup>

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`