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

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.