Skip to content
BeeAI Framework

ACP

Agent Communication Protocol (ACP) is a standard for agent-to-agent communication, allowing different AI agents to interact regardless of how they’re built. This agent works with any ACP-compliant service.

Agent prerequisites

Section titled “Agent prerequisites”
  • Agent Stack installed and running locally
  • BeeAI Framework installed with pip install beeai-framework
  • BeeAI Framework extension for ACP installed with pip install 'beeai-framework[acp]'

ACP Agent

Section titled “ACP Agent”

When to use ACP instead of Agent Stack Integration?

  • You’re connecting to your own custom ACP server
  • You’re developing a multi-agent system where agents communicate via ACP
  • You’re integrating with a third-party ACP-compliant service that isn’t the Agent Stack
import asyncio
import sys
import traceback


from beeai_framework.adapters.acp.agents import ACPAgent
from beeai_framework.errors import FrameworkError
from beeai_framework.memory.unconstrained_memory import UnconstrainedMemory
from examples.helpers.io import ConsoleReader




async def main() -> None:
    reader = ConsoleReader()


    agent = ACPAgent(agent_name="chat", url="http://127.0.0.1:8001", memory=UnconstrainedMemory())
    for prompt in reader:
        # Run the agent and observe events
        response = await agent.run(prompt).on(
            "update",
            lambda data, event: (reader.write("Agent 🤖 (debug) : ", data)),
        )


        reader.write("Agent 🤖 : ", response.last_message.text)




if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

The availability of ACP agents depends on the server you’re connecting to. You can check which agents are available by using the check_agent_exists method:

try:
  await agent.check_agent_exists()
  print("Agent exists and is available")
except AgentError as e:
  print(f"Agent not available: {e.message}")

If you need to create your own ACP server with custom agents, BeeAI framework provides the AcpServer class.

ACP Server

Section titled “ACP Server”

Basic example:

from beeai_framework.adapters.acp import ACPServer, ACPServerConfig
from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.backend import ChatModel
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.weather import OpenMeteoTool




def main() -> None:
    llm = ChatModel.from_name("ollama:granite4:micro")
    agent = RequirementAgent(
        llm=llm,
        tools=[DuckDuckGoSearchTool(), OpenMeteoTool()],
        memory=UnconstrainedMemory(),
        # specify the agent's name and other metadata
        name="chat",
        description="A simple agent",
    )


    # Register the agent with the ACP server and run the HTTP server
    # For the ToolCallingAgent and ReActAgent, we don't need to specify ACPAgent factory method
    # because they are already registered in the ACPServer
    ACPServer(config=ACPServerConfig(port=8001)).register(agent, tags=["example"]).serve()




if __name__ == "__main__":
    main()

Agent execution (Python)

Pass execution=AgentExecutionConfig(...) to server.register() to configure max_iterations, total_max_retries, and max_retries_per_step for the built-in RequirementAgent, ToolCallingAgent, and ReActAgent. These options apply to every request served by that registration. Omitted fields and fields set to None retain the agent’s own defaults. The configuration is copied during registration, and non-empty execution configuration is rejected for other runnable types.

Custom agent example:

import sys
import traceback
from collections.abc import AsyncGenerator
from typing import Unpack


import acp_sdk.models as acp_models
import acp_sdk.server.context as acp_context
import acp_sdk.server.types as acp_types


from beeai_framework.adapters.acp import ACPServer
from beeai_framework.adapters.acp.serve._utils import acp_msgs_to_framework_msgs
from beeai_framework.adapters.acp.serve.agent import ACPServerAgent
from beeai_framework.adapters.acp.serve.server import ACPServerMetadata, to_acp_agent_metadata
from beeai_framework.agents import AgentOptions, AgentOutput, BaseAgent
from beeai_framework.backend.message import AnyMessage, AssistantMessage, UserMessage
from beeai_framework.emitter.emitter import Emitter
from beeai_framework.errors import FrameworkError
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.memory.base_memory import BaseMemory
from beeai_framework.runnable import runnable_entry




# This is a simple echo agent that echoes back the last message it received.
class EchoAgent(BaseAgent):
    memory: BaseMemory


    def __init__(self, memory: BaseMemory) -> None:
        super().__init__()
        self.memory = memory


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


    @runnable_entry
    async def run(self, input: str | list[AnyMessage], /, **kwargs: Unpack[AgentOptions]) -> AgentOutput:
        assert self.memory is not None


        if isinstance(input, str):
            await self.memory.add(UserMessage(input))
        elif isinstance(input, list):
            await self.memory.add_many(input)


        text_input = self.memory.messages[-1].text if self.memory.messages else ""
        return AgentOutput(output=[AssistantMessage(text_input)])




def main() -> None:
    # Create a custom agent factory for the EchoAgent
    def agent_factory(agent: EchoAgent, *, metadata: ACPServerMetadata | None = None) -> ACPServerAgent:
        """Factory method to create an ACPAgent from a EchoAgent."""
        if metadata is None:
            metadata = {}


        async def run(
            input: list[acp_models.Message], context: acp_context.Context
        ) -> AsyncGenerator[acp_types.RunYield, acp_types.RunYieldResume]:
            framework_messages = acp_msgs_to_framework_msgs(input)
            response = await agent.run(framework_messages)
            yield acp_models.MessagePart(content=response.last_message.text, role="assistant")  # type: ignore[call-arg]


        # Create an ACPAgent instance with the run function
        return ACPServerAgent(
            fn=run,
            name=metadata.get("name", agent.meta.name),
            description=metadata.get("description", agent.meta.description),
            metadata=to_acp_agent_metadata(metadata),
        )


    # Register the custom agent factory with the ACP server
    # pyrefly: ignore [bad-argument-type]
    ACPServer.register_factory(EchoAgent, agent_factory)
    # Create an instance of the EchoAgent with UnconstrainedMemory
    agent = EchoAgent(memory=UnconstrainedMemory())
    # Register the agent with the ACP server and run the HTTP server
    ACPServer().register(agent, name="echo_agent").serve()




if __name__ == "__main__":
    try:
        main()
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())


# run: beeai agent run echo_agent "Hello"

Source: python/examples/serve/acp_with_custom_agent.py