Connecting LangGraph agents over A2A

LangGraph makes it easy to compose single-agent workflows, but production systems often need independent agents running as separate processes. This tutorial walks through a concrete langgraph a2a integration: we stand up two LangGraph agents that talk over a minimal Agent-to-Agent (A2A) HTTP contract. By the end you’ll have a researcher agent exposed as a service and a writer agent that calls it over the network.

Prerequisites

• Python 3.11+
• langgraph and langchain-openai (LangGraph 0.2.x)
• fastapi and uvicorn for the A2A server
• httpx for the client-side tool call
• An OpenAI-compatible LLM endpoint and API key

Install them:

pip install langgraph langchain-openai fastapi uvicorn httpx

You should be comfortable with Python async basics and LangGraph’s StateGraph primitive. The langgraph a2a integration we build uses a plain JSON envelope, so no custom protobuf or broker is required.

Architecture of the integration

We define two agents:

1. Researcher — a LangGraph app that answers factual queries. It runs inside a FastAPI process and listens on POST /a2a.
2. Writer — a LangGraph app that uses a tool to call the Researcher over HTTP, then drafts a short brief.

The A2A message format is deliberately boring:

{
  "sender": "writer",
  "message": "What is the current standard for JSON schema draft?"
}

The response mirrors the shape. This keeps both sides decoupled and testable.

Step 1: Implement the researcher agent

The researcher is a single-node LangGraph graph. For LLM calls, any OpenAI-compatible endpoint works; if you want one gateway that addresses 240+ models with automatic fallback when a provider is degraded, point ChatOpenAI at n4n.ai’s OpenAI-compatible endpoint.

# researcher_agent.py
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, BaseMessage

class State(TypedDict):
    messages: Annotated[list[BaseMessage], "conversation"]

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    # base_url="https://api.n4n.ai/v1",  # optional gateway
    # api_key="sk-...",
)

def research_node(state: State) -> State:
    sys = SystemMessage(content="You are a concise researcher. Answer in one paragraph.")
    response = llm.invoke([sys] + state["messages"])
    return {"messages": state["messages"] + [response]}

builder = StateGraph(State)
builder.add_node("research", research_node)
builder.set_entry_point("research")
builder.add_edge("research", END)
researcher_app = builder.compile()

This compiles to an app with .invoke({"messages": [HumanMessage(content="...")]}).

Step 2: Expose the agent via an A2A HTTP endpoint

Wrap the compiled graph in FastAPI. Validate the inbound envelope with Pydantic.

# a2a_server.py
from fastapi import FastAPI
from pydantic import BaseModel
from researcher_agent import researcher_app
from langchain_core.messages import HumanMessage

app = FastAPI()

class A2AMessage(BaseModel):
    sender: str
    message: str

@app.post("/a2a")
def handle_a2a(msg: A2AMessage) -> dict:
    result = researcher_app.invoke({"messages": [HumanMessage(content=msg.message)]})
    reply = result["messages"][-1].content
    return {"sender": "researcher", "message": reply}

Run it:

uvicorn a2a_server:app --port 8000

Expected health check via curl:

curl -s -X POST http://localhost:8000/a2a \
  -H 'content-type: application/json' \
  -d '{"sender":"test","message":"What is A2A?"}'

Sample response:

{"sender":"researcher","message":"A2A commonly refers to agent-to-agent communication protocols that let independent software agents exchange messages and delegate tasks over a network."}

Step 3: Define the writer agent and its A2A tool

The writer needs a LangChain tool that performs the HTTP call. Keep the timeout explicit—cross-process calls fail in ways in-process calls don’t.

# writer_tools.py
import httpx
from langchain_core.tools import tool

@tool
def call_researcher(query: str) -> str:
    """Send a query to the researcher agent over A2A and return its answer."""
    payload = {"sender": "writer", "message": query}
    try:
        resp = httpx.post("http://localhost:8000/a2a", json=payload, timeout=30.0)
        resp.raise_for_status()
    except httpx.HTTPError as e:
        return f"A2A call failed: {e}"
    return resp.json()["message"]

Step 4: Compose the writer graph

The writer graph uses a tool-calling LLM node. We bind the A2A tool and let LangGraph handle the cycle.

# writer_agent.py
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, BaseMessage
from writer_tools import call_researcher

class WState(TypedDict):
    messages: Annotated[list[BaseMessage], "conv"]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools([call_researcher])

def writer_node(state: WState) -> WState:
    sys = SystemMessage(content="You are a writer. Use the researcher to gather facts, then write a 2-sentence brief.")
    response = llm_with_tools.invoke([sys] + state["messages"])
    return {"messages": state["messages"] + [response]}

tool_node = ToolNode([call_researcher])

def should_continue(state: WState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

builder = StateGraph(WState)
builder.add_node("writer", writer_node)
builder.add_node("tools", tool_node)
builder.set_entry_point("writer")
builder.add_conditional_edges("writer", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "writer")
writer_app = builder.compile()

This loop continues until the model produces a final answer without tool calls.

Step 5: Run both services and test

Start the researcher server (already on port 8000). In another shell, drive the writer:

# run_writer.py
from writer_agent import writer_app
from langchain_core.messages import HumanMessage

result = writer_app.invoke(
    {"messages": [HumanMessage(content="Write a brief on agent-to-agent protocols.")]}
)
print(result["messages"][-1].content)

Run it:

python run_writer.py

Expected output (abridged):

Researchers describe A2A as a pattern where autonomous agents exchange structured messages to delegate subtasks. The langgraph a2a integration we just built uses a minimal HTTP JSON envelope, making it trivial to scale each agent independently.

If you watch the researcher’s uvicorn logs, you’ll see the inbound POST /a2a with the writer’s query. That round-trip is the core of the langgraph a2a integration.

Step 6: Hardening the integration

The toy above omits concerns you’ll hit in production:

• Auth: put a shared secret in A2AMessage or use a signed header. FastAPI’s Depends makes this straightforward.
• Schema drift: define the A2A envelope in a shared Python package or OpenAPI spec. Both agents should validate against it.
• Timeouts and retries: httpx with a RetryTransport or tenacity wrapper prevents cascading stalls when the researcher is busy.
• Observability: emit trace IDs in the sender field or a separate header so you can correlate calls across processes.
• Model routing: if you swap the researcher’s LLM per request, forward provider cache-control hints and honor client routing directives at the gateway layer.

A subtle bug source: LangGraph’s ToolNode will loop forever if the tool returns an error string but the model keeps calling it. Cap recursion_limit when invoking:

writer_app.invoke({"messages": [...]}, {"recursion_limit": 5})

Wrapping up

You now have a working langgraph a2a integration with two processes, a clean JSON contract, and runnable code at each step. The same pattern scales to three or thirty agents—just register more /a2a endpoints and bind them as tools. Keep the envelope stable, treat network calls as fallible, and the agents will compose cleanly.