A2A endpoint in Agent Server

"supportedInterfaces": [
  {
    "url": "https://your-deployment/a2a/{assistant_id}",
    "protocolBinding": "JSONRPC",
    "protocolVersion": "1.0"
  }
]

{
  "jsonrpc": "2.0",
  "id": "1",
  "method": "SendMessage",
  "params": {
    "message": {
      "role": "ROLE_USER",
      "parts": [{"text": "Second request"}],
      "messageId": "message-2",
      "contextId": "8b1f0e5c-9a3d-4f27-b0c8-2e6a5d4c1b7a"
    },
    "configuration": {"historyScope": "task"}
  }
}

{"jsonrpc": "2.0", "id": "2", "method": "GetTask",
 "params": {"id": "<taskId>", "historyScope": "task"}}

LangGraph-A2A-History-Scope: task

{
  "uri": "https://langchain.com/a2a/extensions/history-scope/v1",
  "description": "Choose task-only or full-context response history",
  "required": false,
  "params": {
    "header": "LangGraph-A2A-History-Scope",
    "values": ["context", "task"],
    "default": "context",
    "methods": ["SendMessage", "GetTask", "ListTasks"]
  }
}

GET /.well-known/agent-card.json?assistant_id={assistant_id}

{
  "metadata": {
    "a2a": {
      "input_modes": ["text/plain", "application/pdf"],
      "output_modes": ["text/plain", "application/pdf"]
    }
  }
}

{ "metadata": { "a2a": { "a2ui": true } } }

"capabilities": {
  "extensions": [
    {
      "uri": "https://a2ui.org/a2a-extension/a2ui/v0.9",
      "description": "Ability to render A2UI v0.9",
      "required": false,
      "params": {
        "v0.9": {
          "supportedCatalogIds": [],
          "acceptsInlineCatalogs": false
        }
      }
    }
  ]
}

A2A_ALLOWED_TOOL_CALL_RESULTS=generative_ui_tool,another_tool

pip install "langgraph-api>=0.13.0"

"""LangGraph A2A conversational agent.

Supports the A2A protocol with messages input for conversational interactions.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any, Dict, List, TypedDict

from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
from openai import AsyncOpenAI


class Context(TypedDict):
    """Context parameters for the agent."""
    my_configurable_param: str


@dataclass
class State:
    """Input state for the agent.

    Defines the initial structure for A2A conversational messages.
    """
    messages: List[Dict[str, Any]]


async def call_model(state: State, runtime: Runtime[Context]) -> Dict[str, Any]:
    """Process conversational messages and returns output using OpenAI."""
    # Initialize OpenAI client
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    # Process the incoming messages
    latest_message = state.messages[-1] if state.messages else {}
    user_content = latest_message.get("content", "No message content")

    # Create messages for OpenAI API
    openai_messages = [
        {
            "role": "system",
            "content": "You are a helpful conversational agent. Keep responses brief and engaging."
        },
        {
            "role": "user",
            "content": user_content
        }
    ]

    try:
        # Make OpenAI API call
        response = await client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=openai_messages,
            max_tokens=100,
            temperature=0.7
        )

        ai_response = response.choices[0].message.content

    except Exception as e:
        ai_response = f"I received your message but had trouble processing it. Error: {str(e)[:50]}..."

    # Create a response message
    response_message = {
        "role": "assistant",
        "content": ai_response
    }

    return {
        "messages": state.messages + [response_message]
    }


# Define the graph
graph = (
    StateGraph(State, context_schema=Context)
    .add_node(call_model)
    .add_edge("__start__", "call_model")
    .compile()
)

#!/usr/bin/env python3
"""Agent-to-Agent conversation simulation using the LangGraph A2A endpoint."""

import asyncio
import aiohttp
import os
import uuid


def extract_text(result: dict) -> str:
    """Best-effort extraction of response text from an A2A result."""
    if "error" in result:
        raise RuntimeError(f"A2A error {result['error']['code']}: {result['error']['message']}")

    for art in result.get("result", {}).get("artifacts", []) or []:
        for part in art.get("parts", []) or []:
            if part.get("kind") == "text" and part.get("text"):
                return part["text"]

    msg = (result.get("result", {}).get("status", {}) or {}).get("message", {}) or {}
    for part in msg.get("parts", []) or []:
        if part.get("kind") == "text" and part.get("text"):
            return part["text"]

    return "(no text found)"


async def send_message(session, port, assistant_id, text, context_id=None):
    """Send an A2A message. Returns (response_text, returned_context_id)."""
    url = f"http://127.0.0.1:{port}/a2a/{assistant_id}"

    message = {
        "role": "user",
        "parts": [{"kind": "text", "text": text}],
        "messageId": str(uuid.uuid4()),
    }

    # A2A multi-turn continuity: reuse contextId across turns and agents.
    # Do not reuse taskId — each turn starts a new task within the same context.
    if context_id:
        message["contextId"] = context_id

    payload = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "message/send",
        "params": {"message": message},
    }

    headers = {"Accept": "application/json"}
    async with session.post(url, json=payload, headers=headers) as response:
        result = await response.json()

    text = extract_text(result)
    returned_context_id = result.get("result", {}).get("contextId") or context_id
    return text, returned_context_id


async def simulate_conversation():
    """Simulate a conversation between two agents."""

    #Assistant IDs
    agent_a_id = os.getenv("AGENT_A_ID")
    agent_b_id = os.getenv("AGENT_B_ID")

    if not agent_a_id or not agent_b_id:
        print("Set AGENT_A_ID and AGENT_B_ID environment variables")
        return

    message = "Hello! Let's have a conversation."
    context_id = None

    async with aiohttp.ClientSession() as session:
        for i in range(3):
            print(f"--- Round {i + 1} ---")

            message, context_id = await send_message(
                session, 2024, agent_a_id, message, context_id=context_id
            )
            print(f"🔵 Agent A: {message}")

            message, context_id = await send_message(
                session, 2025, agent_b_id, message, context_id=context_id
            )
            print(f"🔴 Agent B: {message}\n")


if __name__ == "__main__":
    asyncio.run(simulate_conversation())

import asyncio
import aiohttp
import uuid


async def send_message(session, url, text, context_id=None):
    """Send an A2A message and return (response_text, context_id)."""

    # --- 1. Build the message ---
    # On follow-up turns, include contextId inside the message object so the server
    # associates them with the ongoing conversation. Do not resend taskId: each turn
    # opens a new task within that conversation.
    message = {
        "role": "user",
        "parts": [{"kind": "text", "text": text}],
        "messageId": str(uuid.uuid4()),
    }
    if context_id:
        message["contextId"] = context_id

    # --- 2. Send it ---
    # contextId travels inside the message. Agent Server turns it into the
    # LangGraph thread_id, so no separate tracing field is needed.
    payload = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "message/send",
        "params": {"message": message},
    }

    async with session.post(url, json=payload, headers={"Accept": "application/json"}) as response:
        if response.status != 200:
            raise RuntimeError(f"HTTP {response.status}: {await response.text()}")
        result = await response.json()

    if "error" in result:
        raise RuntimeError(result["error"].get("message", "Unknown error"))

    result_obj = result.get("result", {})
    returned_context_id = result_obj.get("contextId") or context_id
    text_out = next(
        (
            part.get("text", "")
            for art in result_obj.get("artifacts", []) or []
            for part in art.get("parts", []) or []
            if part.get("kind") == "text"
        ),
        "(no text)",
    )
    return text_out, returned_context_id


async def run_conversation(agent_a_url, agent_b_url):
    # --- 3. Share the context across agents ---
    # The first response carries a contextId. Pass it to every agent from then
    # on, and all their traces land in one LangSmith thread.
    context_id = None
    message = "Hello! Let's collaborate."

    async with aiohttp.ClientSession() as session:
        for _ in range(3):
            message, context_id = await send_message(
                session, agent_a_url, message, context_id=context_id
            )

            message, context_id = await send_message(
                session, agent_b_url, message, context_id=context_id
            )


asyncio.run(run_conversation(
    "http://localhost:2024/a2a/<agent_a_assistant_id>",
    "http://localhost:2025/a2a/<agent_b_assistant_id>",
))

from fastapi import FastAPI, Request
from langsmith.integrations.otel import configure as configure_otel
from opentelemetry import trace
import json

# --- 1. Configure OTel ---
# Set up automatic tracing to LangSmith for your non-LangGraph agent.
configure_otel(project_name="my-a2a-project")
tracer = trace.get_tracer(__name__)

app = FastAPI()

@app.middleware("http")
async def set_thread_id_middleware(request: Request, call_next):
    thread_id = None
    if request.method == "POST":
        body_bytes = await request.body()
        if body_bytes:
            # --- 2. Extract contextId from the incoming A2A message ---
            try:
                body = json.loads(body_bytes)
                thread_id = body["params"]["message"].get("contextId")
            except (ValueError, KeyError, TypeError):
                pass
            # Re-inject the body so downstream handlers can still read it
            async def receive():
                return {"type": "http.request", "body": body_bytes}
            request._receive = receive

    # --- 3. Attach thread_id to the trace ---
    # langsmith.metadata.thread_id groups this trace with others in the same thread.
    with tracer.start_as_current_span("agent") as span:
        if thread_id:
            span.set_attribute("langsmith.metadata.thread_id", thread_id)
        return await call_next(request)

curl "https://your-deployment/a2a/{assistant_id}/.well-known/agent-card.json"

curl -X POST "https://your-deployment/a2a/{assistant_id}" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "SendMessage",
    "params": {
      "message": {
        "role": "ROLE_USER",
        "parts": [{"text": "hello"}],
        "messageId": "test-1"
      }
    }
  }'

./run_tck.py --sut-host https://your-deployment/a2a/{assistant_id} --transport jsonrpc

{
  "$schema": "https://langgra.ph/schema.json",
  "http": {
    "disable_a2a": true
  }
}