Supported in Python only.
Prerequisites
- BeeAI Framework installed with
pip install beeai-framework - BeeAI Framework extension for A2A installed with
pip install 'beeai-framework[a2a]'
A2A Client
TheA2AAgent lets you easily connect with external agents using the A2A protocol.
import asyncio
import sys
import traceback
from beeai_framework.adapters.a2a.agents import A2AAgent, A2AAgentUpdateEvent
from beeai_framework.emitter import EventMeta
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 = A2AAgent(url="http://127.0.0.1:9999", memory=UnconstrainedMemory())
for prompt in reader:
# Run the agent and observe events
def print_update(data: A2AAgentUpdateEvent, event: EventMeta) -> None:
value = data.value
debug_info = value[1] if isinstance(value, tuple) else value
reader.write("Agent 🤖 (debug) : ", str(debug_info))
response = await agent.run(prompt).on("update", print_update)
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())
import "dotenv/config.js";
import { A2AAgent } from "beeai-framework/adapters/a2a/agents/agent";
import { createConsoleReader } from "examples/helpers/io.js";
import { FrameworkError } from "beeai-framework/errors";
import { TokenMemory } from "beeai-framework/memory/tokenMemory";
const agent = new A2AAgent({
url: "http://127.0.0.1:9999",
memory: new TokenMemory(),
});
const reader = createConsoleReader();
try {
for await (const { prompt } of reader) {
const result = await agent.run({ input: prompt }).observe((emitter) => {
emitter.on("update", (data) => {
reader.write(`Agent (received progress) 🤖 : `, JSON.stringify(data.value, null, 2));
});
emitter.on("error", (data) => {
reader.write(`Agent (error) 🤖 : `, data.message);
});
});
reader.write(`Agent 🤖 : `, result.result.text);
}
} catch (error) {
reader.write("Agent (error) 🤖", FrameworkError.ensure(error).dump());
}
A2A Server
A2AServer lets you expose agents built in the BeeAI framework via A2A protocol.
A2A supports only one agent per server.
# pyrefly: ignore [missing-module-attribute]
from beeai_framework.adapters.a2a import A2AServer, A2AServerConfig
from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.backend import ChatModel
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.serve.utils import LRUMemoryManager
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(),
)
# Register the agent with the A2A server and run the HTTP server
# For the ToolCallingAgent, we don't need to specify A2AAgent factory method
# because it is already registered in the A2AServer
# we use LRU memory manager to keep limited amount of sessions in the memory
A2AServer(
config=A2AServerConfig(port=9999, protocol="jsonrpc"), memory_manager=LRUMemoryManager(maxsize=100)
).register(agent, send_trajectory=True).serve()
if __name__ == "__main__":
main()
import "dotenv/config.js";
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";
import { ToolCallingAgent } from "beeai-framework/agents/toolCalling/agent";
import { UnconstrainedMemory } from "beeai-framework/memory/unconstrainedMemory";
import { A2AServer } from "beeai-framework/adapters/a2a/serve/server";
// ensure the model is pulled before running
const llm = new OllamaChatModel("granite4:micro");
const agent = new ToolCallingAgent({
llm,
memory: new UnconstrainedMemory(),
tools: [
new OpenMeteoTool(), // weather tool
],
});
await new A2AServer().register(agent).serve();
