astromesh-a2a

Let your agent talk to other agents without trusting them.

Leer en español

A2A (Agent2Agent) is the open protocol agents use to work with each other: an agent publishes a card that says what it can do, and other agents send it tasks over JSON-RPC, get answers, stream progress and answer its questions.

astromesh-a2a is a client and a server for A2A 1.x, built on the official A2A SDK. It puts a remote agent inside your agent framework as an ordinary tool (nine frameworks), serves your own agent as an A2A agent, and treats everything that crosses the wire as what it is: code you do not control.

Why the SDK alone is not enough

The SDK speaks the protocol. It does not decide which remotes you may reach, how much of their output you accept, or what your model does with it. A remote agent is an untrusted program on someone else's server; this library is the layer between it and your agent.

These are not claims: an attack eval runs nine malicious remotes against the library on every push.

Install

pip install astromesh-a2a                # client
pip install 'astromesh-a2a[server]'      # + serve your own agent (Starlette, uvicorn)
pip install 'astromesh-a2a[langchain]'   # + a remote agent as a LangChain tool

Python 3.10+. Every framework is an optional extra (table below); signing adds PyJWT for signed cards, all installs everything.

Call a remote agent

from astromesh_a2a import Auth, Policy, RemoteAgent

policy, auth = Policy(allowed=("trips.example.com",)), Auth.bearer("TRIPS_TOKEN")
async with RemoteAgent("https://trips.example.com", policy=policy, auth=auth) as trips:
    result = await trips.send("Find a flight to Lima on Friday")
print(result.status, result.answer, result.trace)

result.status is completed, or input_required with the remote's question in result.pending; reply(task_id, text) answers it, stream(text) yields events, cancel(task_id) stops a task, card() returns the vetted card. Errors are typed (PolicyRejected, BlockedDestination, TaskFailed, A2ATimeout, UnsupportedVersion, all A2AError). Guide: docs/client.md.

Serve your agent

from astromesh_a2a import CardSpec, Incoming, Outgoing, Skill, api_key_auth, serve


async def handle(msg: Incoming) -> Outgoing:  # msg.text came from another program: untrusted
    return Outgoing(answer=await my_agent.run(msg.text, session=msg.session_key))


skill = Skill(id="book", name="Book trips", description="Flights and hotels.", tags=("travel",))
card = CardSpec("trips", "Books trips.", "1.0.0", "https://trips.example.com", [skill])
serve(handle, card, host="0.0.0.0", port=8000, auth=api_key_auth("TRIPS_API_KEY"))

The card is served at /.well-known/agent-card.json and JSON-RPC (blocking and SSE) at /a2a. Auth is mandatory: without auth= the server refuses to start unless you pass insecure=True. create_app(...) returns the Starlette app to mount in your own server. Guide: docs/server.md.

Try both ends on your machine, offline:

git clone https://github.com/monaccode/astromesh-a2a && cd astromesh-a2a
uv run python examples/hello_local.py

It runs from a checkout (examples are not in the wheel). The egress guard refuses 127.0.0.1, so the example opts its own agent into loopback with astromesh_a2a.testing.allow_loopback(agent): local development and tests only, never production (docs/policy.md).

A remote agent as a tool of your framework

async with RemoteAgent("https://trips.example.com", policy=policy) as trips:
    await trips.card()  # the tool's name and description come from the card
    tool = trips.as_tool("langchain")  # give it to your agent: tools=[tool]
    print(await tool.ainvoke({"query": "Find a flight to Lima on Friday"}))

The tool takes one query, keeps the conversation (and a task waiting for input) on the tool object, returns the wrapped untrusted output and reports errors the framework's way.

The other way round, adapters.server_langchain.handler_from_runnable(runnable) serves a LangChain runnable or LangGraph graph as an A2A agent. One example per framework: docs/frameworks.md.

Sync and async: event loops

A RemoteAgent is bound to the event loop of its first I/O (its pooled connections belong to it) and refuses any other loop with LoopBindingError. Async tools just work. Sync tools (CrewAI run, smolagents, LangChain invoke...) are dispatched to the agent's loop when it runs in another thread, or run on a background loop when the agent has none yet. Do not block the agent's loop while its tools run: calling a sync entry point (crew.kickoff()) inside async def main would deadlock, so the dispatch gives up after policy.timeout + 5 s with LoopBindingError. Use the async entry point (kickoff_async, ainvoke) or await asyncio.to_thread(crew.kickoff). From sync code, use run_sync(agent.card(), agent=agent) and run_sync(agent.aclose(), agent=agent) (from astromesh_a2a.adapters import run_sync). One RemoteAgent per loop.

Sessions

Incoming.context_id is chosen by the caller and is not scoped: two callers can send the same one. Key any memory or session state by Incoming.session_key ("<caller>:<context_id>", where the caller is the identity your auth returns), never by context_id alone. On the client side, pass context_id= to send() to keep several messages in one remote conversation.

The eval

Nine local malicious remotes: a spoofed card, a redirect to an internal address, a public name that resolves to 10.0.0.7, a 32 MiB artifact, a file part, prompt injection that tries to break out of the untrusted block, a remote that asks to be confirmed, a stream that never ends and a 0.3-only agent.

python -m astromesh_a2a.eval   # needs the server extra

The other seven hold whatever the policy says: egress guard, card checks, the untrusted wrapper and the no-auto-confirm rule. CI fails if a change lets one trap through. Traps, "stopped" rules and negative controls: docs/eval.md.

Policy

from astromesh_a2a import Budget, Policy

policy = Policy(
    allowed=("trips.example.com", "https://hotels.example.org/.well-known/agent-card.json"),
    timeout=30.0,
    budget=Budget(max_calls=20, max_seconds=120),
)

Every field, its default and why: docs/policy.md.

In the Astromesh ecosystem

astromesh-a2a is the A2A layer of Astromesh, the same way astromesh-rag is its RAG layer and astromesh-okf its knowledge format. Planned (not shipped yet):

• Astromesh core: a type: a2a tool in the agent manifest (card_url, auth_env, policy, timeout) built on this library, with input_required returned to the model as an observation; turns from remote agents counted as agent turns, never human ones.
• Astromesh ADK: remote_a2a() to use a remote A2A agent from Python agents.
• Nexus: published agents served over A2A (opt-in per agent), conformance-tested against this client.

Details: docs/ecosystem.md.

Contributing

Issues and pull requests are welcome: CONTRIBUTING.md.

License

Apache-2.0. Part of the Astromesh ecosystem.