langchain-remote-a2a-agent 1.0.0

LangChain-native equivalent of Google ADK's RemoteA2aAgent — invoke remote A2A agents as LangChain Runnables and tools

langchain-remote-a2a-agent

LangChain-native equivalent of Google ADK's RemoteA2aAgent. Wraps any Agent2Agent (A2A) protocol server as a LangChain Runnable, so a remote agent can be .invoke()d, .astream()ed, dropped into a create_agent/LangGraph pipeline as a node, or exposed to another agent as a callable tool via .as_tool() — the same interface you'd use for any local LangChain component.

This is useful when the agent doing the actual work runs as a separate service (different language, different team, different deployment) and you want to compose it into a LangChain-orchestrated system without hand-rolling the A2A JSON-RPC/SSE client plumbing yourself.

Install

pip install langchain-remote-a2a-agent

Requires Python 3.11+. Ships a py.typed marker — type checkers (mypy, Pyright) and IDE autocomplete/hover (VS Code + Pylance) pick up the inline type hints and docstrings automatically once installed.

Quickstart

from langchain_remote_a2a_agent import RemoteAgent

research = RemoteAgent(
    name="research",
    description="Research agent",
    agent_card_url="http://localhost:8001/.well-known/agent-card.json",
)

result = research.invoke("What is the boiling point of nitrogen?")
print(result["messages"][-1].content)

Multi-turn conversations

Pass a thread_id to resume the same server-side A2A context across calls:

from langchain_core.runnables import RunnableConfig

config = RunnableConfig(configurable={"thread_id": "user-42"})

result1 = research.invoke("Tell me about black holes.", config)
result2 = research.invoke("How do they emit Hawking radiation?", config)

Streaming

async for chunk in research.astream("Explain quantum entanglement."):
    if chunk.get("messages"):
        msg = chunk["messages"][-1]
        if reasoning := msg.additional_kwargs.get("reasoning_content"):
            print(f"[reasoning] {reasoning}")
        elif msg.content:
            print(msg.content, end="", flush=True)

As a tool inside a LangChain agent

from langchain.agents import create_agent

supervisor = create_agent(
    model="openai:gpt-5",
    tools=[research.as_tool()],
)

More examples

Nine complete, runnable examples — synchronous/async invocation, multi-turn state, streaming with reasoning traces, create_agent integration, LangGraph node usage, concurrent batching, bearer-token auth, and agent-card inspection — live in langchain_remote_a2a_agent/examples.py. They're also directly importable for reference:

from langchain_remote_a2a_agent import examples

API surface

Export	What it is
RemoteAgent	The main Runnable — wraps a remote A2A agent.
RemoteAgent.as_tool()	Returns a BaseTool for use inside another agent's tool list.
RemoteAgent.reset_thread()	Clears the stored context_id for a thread_id, starting a fresh server-side conversation.
RemoteAgent.thread_state()	JSON-serialisable snapshot of a thread's tracked state.
RemoteAgent.aget_agent_card()	Fetch (or return cached) the remote agent's Agent Card.
RemoteAgent.aget_extended_agent_card()	Fetch the remote agent's Extended Agent Card.
ThreadState, ConversationStateStore	The conversation-state tracking types, exposed for custom store implementations.
RemoteAgentError and subclasses	CardResolutionError, A2AProtocolError, A2ATimeoutError, A2AAuthError, A2AStreamError, InputNormalisationError — catch these individually or RemoteAgentError for all of them.

Agent Card Inspection

You can inspect the capabilities, skills, security requirements, and metadata of the remote agent by fetching its Agent Card:

Public Agent Card

The Public Agent Card is resolved from the agent card URL when the agent is first initialized:

# Fetch (or return cached) the public agent card
card = await research.aget_agent_card()

print("Agent name:", card.name)
print("Description:", card.description)
print("Skills:", card.skills)

Extended Agent Card

If the remote agent supports an extended card (carrying additional metadata or developer-defined parameters that might require auth/permissions), you can fetch it using aget_extended_agent_card():

# Fetch the extended agent card from the remote server
extended_card = await research.aget_extended_agent_card()

print("Developer info:", extended_card.provider.organization)
print("Extended capabilities:", extended_card.capabilities)

Github Repo

LangChain Remote A2A Agent

Tests

# Run against a live agent
pytest tests/test_integration.py -v --agent-url http://localhost:8001/.well-known/agent-card.json

# With auth token for extended card
pytest tests/test_integration.py -v \
  --agent-url http://localhost:8001/.well-known/agent-card.json \
  --auth-token "Bearer sk-..."

# Custom test query
pytest tests/test_integration.py -v \
  --agent-url http://localhost:8001/.well-known/agent-card.json \
  --test-query "Summarize quantum computing"

License

Apache License 2.0 — see LICENSE.