A2A

Splunk Agent Observability traces A2A (Agent2Agent) interactions using the splunk-ao-a2a instrumentor — giving you a single distributed trace across agents, including LLM calls, tool use, and cross-agent handoffs.

Setup

Installation

Add the OpenTelemetry packages to your project:

pip install opentelemetry-api opentelemetry-sdk \
            opentelemetry-exporter-otlp

The opentelemetry-api and opentelemetry-sdk packages provide the core OpenTelemetry functionality. The opentelemetry-exporter-otlp package enables sending traces to Splunk Agent Observability’s OTLP endpoint.

Create environment variables for your Splunk Agent Observability settings

Set environment variables for your Splunk Agent Observability settings, for example in a .env file. These environment variables are consumed by the SplunkAOSpanProcessorto authenticate and route traces to the correct Splunk Agent Observability Project and Agent Stream:

# Provide your Splunk Agent Observability
# API key and URL if you are using an
# on-premises, standalone, or custom deployment
# SPLUNK_AO_API_KEY="your-splunk-ao-api-key"
# SPLUNK_AO_CONSOLE_URL="your-splunk-ao-url"

# Your Splunk Agent Observability project name
SPLUNK_AO_PROJECT="your-splunk-ao-project-name"

# The name of the Agent Stream you want to use for logging
SPLUNK_AO_AGENT_STREAM="your-splunk-ao-agent-stream"

Self hosted deployments: Set the OTel endpoint

Skip this step if you are using a hosted version.

The OTel endpoint is different from Splunk Agent Observability’s regular API endpoint and is specifically designed to receive telemetry data in the OTLP format.If you are using:

A cloud deployment, then you don’t need to provide a custom OTel endpoint. The default endpoint <your-splunk-ao-api-url>/otel/traces will be used automatically.
A self-hosted deployment, replace the <your-splunk-ao-api-url>/otel/traces endpoint with your deployment URL. The format of this URL is based on your console URL, appending /otel/traces.

The convention is to store this in the SPLUNK_AO_CONSOLE_URL environment variable. For example:

os.environ["SPLUNK_AO_CONSOLE_URL"] = "<your-splunk-ao-url>"

Initialize and create the Splunk Agent Observability span processor

The SplunkAOSpanProcessor automatically configures authentication and metadata using your environment variables. It also:

Auto-builds OTLP headers using your Splunk Agent Observability credentials
Configures the correct OTLP trace endpoint
Registers a batch span processor that exports traces to Splunk Agent Observability

from splunk_ao import otel  

# SplunkAOSpanProcessor (no manual OTLP config required) loads the env vars for 
# the Splunk Agent Observability API key, Project, and Agent Stream. Make sure to set them first. 
splunk_ao_span_processor = otel.SplunkAOSpanProcessor(
    # Optional parameters if not set, uses env var
    # project=os.environ["SPLUNK_AO_PROJECT"], 
    # agentstream=os.environ.get("SPLUNK_AO_AGENT_STREAM"),  
)

Install and instrument A2A

pip install splunk-ao-a2a opentelemetry-instrumentation-langchain

from splunk_ao_a2a import A2AInstrumentor
from opentelemetry.instrumentation.langchain import LangchainInstrumentor

# Trace A2A protocol
A2AInstrumentor().instrument(tracer_provider=provider, agent_name="my-agent")

# Trace LangChain/LangGraph agents, LLM calls, and tools
LangchainInstrumentor().instrument(tracer_provider=provider)

All a2a-sdk client and server methods are now traced automatically.

How it works

When Agent A calls Agent B, trace context is propagated through A2A message metadata. Both agents’ spans appear in one distributed trace:

Agent A: LangGraph orchestrator             Agent B: LangChain researcher
+-----------------------------------+       +------------------------------+
| invoke_agent LangGraph            |       | a2a.server.on_message_send   |
|  +-- plan (LLM)                   |       |  +-- invoke_agent LangGraph  |
|  +-- delegate                     |  a2a  |     +-- LLM + tool_calls     |
|  |    +-- a2a.client.send_message |-----> |     +-- search_kb (tool)     |
|  +-- synthesize (LLM)             |  ctx  |     +-- LLM final answer     |
+-----------------------------------+       +------------------------------+
                            Single trace in Splunk Agent Observability

A2A context_id is mapped to session.id, grouping all interactions in the same conversation into one Splunk Agent Observability session.

Full example

Two LangGraph agents in one script. Agent A (orchestrator) plans, delegates to Agent B via A2A, then synthesizes. Agent B (researcher) uses a tool and LLM to answer. Copy-paste and run.

pip install splunk-ao-a2a "splunk-ao[otel]" langchain langchain-openai langgraph \
    opentelemetry-instrumentation-langchain uvicorn starlette sse-starlette \
    python-dotenv

.env

SPLUNK_AO_PROJECT=your-splunk-ao-project
SPLUNK_AO_AGENT_STREAM=your-agent-stream
OPENAI_API_KEY=your-openai-api-key
# If you are using an on-premises, standalone, or custom deployment
# provide your API key and deployment URL below
# SPLUNK_AO_API_KEY=your-splunk-ao-api-key
 #SPLUNK_AO_CONSOLE_URL=your-splunk-ao-url

import asyncio
import uuid

import httpx
import uvicorn
from a2a.client import ClientConfig, ClientFactory
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.apps.jsonrpc.starlette_app import A2AStarletteApplication
from a2a.server.events import EventQueue, InMemoryQueueManager
from a2a.server.request_handlers.default_request_handler import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities, AgentCard, AgentSkill, Message, Role,
    TaskState, TaskStatus, TaskStatusUpdateEvent, TextPart,
)
from dotenv import load_dotenv
from splunk_ao.otel import SplunkAOSpanProcessor, add_splunk_ao_span_processor
from splunk_ao_a2a import A2AInstrumentor
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from opentelemetry.instrumentation.langchain import LangchainInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from starlette.applications import Starlette
from typing_extensions import TypedDict

# Load SPLUNK_AO_API_KEY, SPLUNK_AO_CONSOLE_URL, OPENAI_API_KEY, etc. from .env
# into the process environment before constructing SplunkAOSpanProcessor.
# SPLUNK_AO_API_KEY is only required for on-premises, standalone, and custom deployments.
load_dotenv()

# --- Splunk Agent Observability tracing setup ---
provider = TracerProvider()
add_splunk_ao_span_processor(provider, SplunkAOSpanProcessor())
A2AInstrumentor().instrument(tracer_provider=provider, agent_name="orchestrator")
LangchainInstrumentor().instrument(tracer_provider=provider)

llm = ChatOpenAI(model="gpt-4o-mini")


# --- Agent B: researcher (served over A2A) ---

@tool
def search_kb(query: str) -> str:
    """Search the travel knowledge base."""
    if "paris" in query.lower():
        return "Eiffel Tower 330m, Louvre 9.6M visitors/yr, 20 arrondissements."
    return f"No results for: {query}"

researcher = create_agent(
    llm, [search_kb],
    system_prompt="Use search_kb to find facts, then summarize for a traveler.",
)

class ResearcherExecutor(AgentExecutor):
    async def execute(self, ctx: RequestContext, queue: EventQueue) -> None:
        result = await researcher.ainvoke({"messages": [("user", ctx.get_user_input())]})
        await queue.enqueue_event(TaskStatusUpdateEvent(
            task_id=ctx.task_id, context_id=ctx.context_id, final=True,
            status=TaskStatus(
                state=TaskState.completed,
                message=Message(
                    message_id=str(uuid.uuid4()), role=Role.agent,
                    parts=[TextPart(text=result["messages"][-1].content or "")],
                ),
            ),
        ))

    async def cancel(self, ctx: RequestContext, queue: EventQueue) -> None:
        await queue.enqueue_event(TaskStatusUpdateEvent(
            task_id=ctx.task_id, context_id=ctx.context_id, final=True,
            status=TaskStatus(state=TaskState.canceled),
        ))

CARD = AgentCard(
    name="researcher",
    description="Travel researcher with tool use",
    url="http://localhost:9867",
    version="1.0.0",
    capabilities=AgentCapabilities(streaming=True),
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    skills=[AgentSkill(id="qa", name="Q&A", description="Answer questions", tags=[])],
)


# --- Agent A: orchestrator (LangGraph) ---

class OrchestratorState(TypedDict):
    user_query: str
    research_query: str
    response: str
    plan: str

def build_orchestrator(client):
    async def plan(state: OrchestratorState) -> dict:
        prompt = (
            "Formulate a travel research question."
            " Reply with ONLY the question."
        )
        result = await create_agent(
            llm, system_prompt=prompt,
        ).ainvoke({"messages": [("user", state["user_query"])]})
        return {"research_query": result["messages"][-1].content}

    async def delegate(state: OrchestratorState) -> dict:
        msg = Message(
            message_id=str(uuid.uuid4()), role=Role.user,
            parts=[TextPart(text=state["research_query"])],
            context_id="session-1",
        )
        async for event in client.send_message(msg):
            if isinstance(event, tuple):
                task = event[0]
                is_done = (
                    task.status
                    and task.status.state == TaskState.completed
                    and task.status.message
                )
                if is_done:
                    text = getattr(
                        task.status.message.parts[0].root,
                        "text", "",
                    )
                    return {"response": text}
        return {"response": ""}

    async def synthesize(state: OrchestratorState) -> dict:
        prompt = "Create a brief 3-day itinerary from the research."
        user_msg = (
            f"Research:\n{state['response']}"
            "\n\nCreate itinerary."
        )
        result = await create_agent(
            llm, system_prompt=prompt,
        ).ainvoke({"messages": [("user", user_msg)]})
        return {"plan": result["messages"][-1].content}

    graph = StateGraph(OrchestratorState)
    graph.add_node("plan", plan)
    graph.add_node("delegate", delegate)
    graph.add_node("synthesize", synthesize)
    graph.add_edge(START, "plan")
    graph.add_edge("plan", "delegate")
    graph.add_edge("delegate", "synthesize")
    graph.add_edge("synthesize", END)
    return graph.compile()


# --- Run both agents ---

async def main():
    # Start Agent B
    app = Starlette()
    A2AStarletteApplication(
        agent_card=CARD,
        http_handler=DefaultRequestHandler(
            agent_executor=ResearcherExecutor(),
            task_store=InMemoryTaskStore(),
            queue_manager=InMemoryQueueManager(),
        ),
    ).add_routes_to_app(app)
    server = uvicorn.Server(uvicorn.Config(app, port=9867, log_level="warning"))
    server_task = asyncio.create_task(server.serve())
    await asyncio.sleep(1)

    # Run Agent A
    client = ClientFactory(
        config=ClientConfig(
            streaming=True,
            httpx_client=httpx.AsyncClient(
                timeout=httpx.Timeout(120),
            ),
        ),
    ).create(CARD)
    result = await build_orchestrator(client).ainvoke({
        "user_query": "Plan a 3-day trip to Paris",
        "research_query": "",
        "response": "",
        "plan": "",
    })
    print(result["plan"])

    server.should_exit = True
    await server_task
    provider.shutdown()

asyncio.run(main())

Overview

Get Started

Observability

Evaluators

Signals

AI Assistant

Luna Studio

Experiments

Agent Control

Annotations

Integrations

Security

References

Setup

How it works

Full example

​Setup

​How it works

​Full example

Setup

How it works

Full example