Agent-to-Agent (A2A)

The A2A protocol is an open standard for agent discovery, communication, and collaboration across platforms and services. Elsai implements A2A so your agents can call remote agents — or expose themselves as — A2A-compatible services.

Installation

pip install --extra-index-url https://elsai-agents.elsai.ai/root/ "elsai-agents[a2a]==0.1.0"

Exposing an agent as an A2A server

from elsai import Agent, tool
from elsai.agent.a2a import A2AServer

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))

agent = Agent(
    name="calculator",
    system_prompt="You are a precise calculator agent.",
    tools=[calculate],
)

server = A2AServer(agent=agent, host="0.0.0.0", port=9000)
server.serve()
# Agent card served at: GET  /.well-known/agent-card.json
# JSON-RPC endpoint at:  POST /

Calling a remote A2A agent

from elsai.agent.a2a import A2AAgent

remote = A2AAgent(
    endpoint="http://localhost:9000",
    name="remote-calculator",
    timeout=300,   # seconds
)

result = remote("What is 10 ^ 6?")
print(result)   # → "10^6 = 1,000,000"

The agent card at /.well-known/agent-card.json is fetched and cached automatically on the first call.

A2A as an orchestrator tool

Wrap a remote agent as a tool for a local orchestrator:

from elsai import Agent
from elsai.agent.a2a import A2AAgent

calculator = A2AAgent(endpoint="http://calc-service:9000")
researcher  = A2AAgent(endpoint="http://research-service:9000")

orchestrator = Agent(
    name="orchestrator",
    system_prompt="Delegate calculations to calculator, research to researcher.",
    tools=[calculator.as_tool(), researcher.as_tool()],
)

result = orchestrator("Research Tokyo's population and calculate its density per km²")
print(result)

A2A as a graph node

Mix local and remote agents inside a Graph:

from elsai import Agent
from elsai.agent.a2a import A2AAgent
from elsai.multiagent import Graph

local_writer   = Agent(name="writer",  system_prompt="Write polished reports.")
remote_analyst = A2AAgent(endpoint="http://analyst-service:9000")

graph = Graph()
graph.add_node(remote_analyst)
graph.add_node(local_writer)
graph.add_edge(remote_analyst, local_writer)

result = graph("Analyse and write a report on Q3 revenue trends")
print(result)

Async and streaming

import asyncio

async def main():
    remote = A2AAgent(endpoint="http://localhost:9000")

    # Async invocation
    result = await remote.invoke_async("Calculate √144")
    print(result)

    # Streaming
    async for event in remote.stream_async("Explain quantum entanglement"):
        if "data" in event:
            print(event["data"], end="")

asyncio.run(main())

Agent Card

Every A2A server advertises capabilities at /.well-known/agent-card.json:

{
  "name": "calculator",
  "description": "A precise calculator agent",
  "skills": ["arithmetic", "algebra", "unit-conversion"],
  "endpoint": "http://localhost:9000"
}

A2AServer options

Parameter	Default	Description
host	127.0.0.1	Bind address
port	9000	Port
http_url	auto-detected	Public URL advertised in the agent card
task_store	In-memory	Persistence backend for task state

A2AAgent options

Parameter	Default	Description
endpoint	required	Remote A2A server URL
name	"a2a-agent"	Name for this client agent
description	""	Description used when wrapping as a tool
timeout	300	Request timeout in seconds

Swarm not supported

A2A is not yet supported inside Swarm patterns. Use Graph or agent-as-tool instead.