Google ADK (Native)

The splunk-ao-adk package provides native observability for Google ADK agents. It automatically traces agent runs, LLM calls, and tool executions with minimal setup.

This is the native integration for Google ADK. If you prefer using OpenTelemetry, see the OpenTelemetry-based integration.

Installation

pip install splunk-ao-adk

View the package on PyPI for version history and additional details.

Requirements: Python 3.10+, a Splunk Agent Observability API key, and a Google AI API key

Quick start

The simplest way to add Splunk Agent Observability to your Google ADK application is using the SplunkAOADKPlugin. This plugin attaches to the ADK Runner and automatically captures all agent, LLM, and tool events.

import asyncio
from splunk_ao_adk import SplunkAOADKPlugin
from google.adk.runners import Runner
from google.adk.agents import LlmAgent
from google.genai import types

async def main():
    # Create the Splunk Agent Observability ADK plugin
    plugin = SplunkAOADKPlugin(project="my-project", agent_stream="production")

    # Create your agent
    agent = LlmAgent(
        name="assistant",
        model="gemini-2.0-flash",
        instruction="You are helpful."
    )

    # Pass the plugin to the Runner
    runner = Runner(agent=agent, plugins=[plugin])

    # Run the agent
    message = types.Content(
        parts=[types.Part(text="Hello! What can you help me with?")]
    )
    async for event in runner.run_async(
        user_id="user-123",
        session_id="session-456",
        new_message=message
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)

if __name__ == "__main__":
    # Set environment variables: GOOGLE_API_KEY and
    # SPLUNK_AO_API_KEY (only for on-premises, standalone, or custom deployments)
    asyncio.run(main())

Once the plugin is configured, all agent runs are automatically logged to Splunk Agent Observability, including:

Agent execution spans
LLM calls with input/output and token usage
Tool executions with arguments and results
Error handling and status codes

Configuration

The plugin can be configured via constructor parameters or environment variables:

Parameter	Environment Variable	Description
`project`	`SPLUNK_AO_PROJECT`	Project name (required unless `ingestion_hook` provided)
`agent_stream`	`SPLUNK_AO_AGENT_STREAM`	Agent Stream name (required unless `ingestion_hook` provided)
`ingestion_hook`	-	Custom callback for trace data (bypasses Splunk Agent Observability backend)

Features

Session tracking

All traces with the same ADK session_id are automatically grouped into a Splunk Agent Observability session. This enables conversation-level tracking across multiple interactions:

import asyncio
from splunk_ao_adk import SplunkAOADKPlugin
from google.adk.runners import Runner
from google.adk.agents import LlmAgent
from google.genai import types

async def main():
    plugin = SplunkAOADKPlugin(project="my-project", agent_stream="production")
    agent = LlmAgent(
        name="assistant",
        model="gemini-2.0-flash",
        instruction="You are helpful."
    )
    runner = Runner(agent=agent, plugins=[plugin])

    # All traces in this conversation are grouped together
    session_id = "conversation-abc"

    # First message
    message1 = types.Content(
        parts=[types.Part(text="Hello! What's the capital of France?")]
    )
    async for event in runner.run_async(
        user_id="user-123",
        session_id=session_id,
        new_message=message1
    ):
        if event.is_final_response():
            print(f"Response 1: {event.content.parts[0].text}")

    # Follow-up in same session
    message2 = types.Content(
        parts=[types.Part(text="What about Germany?")]
    )
    async for event in runner.run_async(
        user_id="user-123",
        session_id=session_id,
        new_message=message2
    ):
        if event.is_final_response():
            print(f"Response 2: {event.content.parts[0].text}")

if __name__ == "__main__":
    # Set environment variables: GOOGLE_API_KEY and
    # SPLUNK_AO_API_KEY (only required for on-premises, standalone, and custom deployments)
    asyncio.run(main())

Custom metadata

Attach custom metadata to traces using ADK’s native RunConfig.custom_metadata. Metadata is propagated to all spans (agent, LLM, tool) within the invocation:

import asyncio
from splunk_ao_adk import SplunkAOADKPlugin
from google.adk.runners import Runner
from google.adk.agents import LlmAgent
from google.adk.agents.run_config import RunConfig
from google.genai import types

async def main():
    plugin = SplunkAOADKPlugin(project="my-project", agent_stream="production")
    agent = LlmAgent(
        name="assistant",
        model="gemini-2.0-flash",
        instruction="You are helpful."
    )
    runner = Runner(agent=agent, plugins=[plugin])

    # Attach custom metadata to the trace
    run_config = RunConfig(
        custom_metadata={
            "user_tier": "premium",
            "conversation_id": "conv-abc",
            "turn": 1,
            "experiment_group": "A",
        }
    )

    message = types.Content(
        parts=[types.Part(text="Hello! Tell me a fun fact.")]
    )
    async for event in runner.run_async(
        user_id="user-123",
        session_id="session-456",
        new_message=message,
        run_config=run_config,  # Pass the run config with metadata
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)

if __name__ == "__main__":
    # Set environment variables: GOOGLE_API_KEY and
    # SPLUNK_AO_API_KEY (only required for on-premises, standalone, and custom deployments)
    asyncio.run(main())

Callback mode

For granular control over which callbacks to use, you can attach them directly to your agent instead of using the plugin. This is useful when you need to customize behavior for specific agents:

import asyncio
from splunk_ao_adk import SplunkAOADKCallback
from google.adk.runners import Runner
from google.adk.agents import LlmAgent
from google.genai import types

async def main():
    # Create the Splunk Agent Observability ADK callback
    callback = SplunkAOADKCallback(project="my-project", agent_stream="production")

    # Attach callbacks directly to your agent
    agent = LlmAgent(
        name="assistant",
        model="gemini-2.0-flash",
        instruction="You are helpful.",
        before_agent_callback=callback.before_agent_callback,
        after_agent_callback=callback.after_agent_callback,
        before_model_callback=callback.before_model_callback,
        after_model_callback=callback.after_model_callback,
        before_tool_callback=callback.before_tool_callback,
        after_tool_callback=callback.after_tool_callback,
    )

    # Create the runner without the plugin
    runner = Runner(agent=agent)

    message = types.Content(
        parts=[types.Part(text="Hello! How are you?")]
    )
    async for event in runner.run_async(
        user_id="user-123",
        session_id="session-456",
        new_message=message
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)

if __name__ == "__main__":
    # Set environment variables: GOOGLE_API_KEY and 
    # SPLUNK_AO_API_KEY (only for on-premises, standalone, and custom deployments)
    asyncio.run(main())

Retriever spans

By default, all FunctionTool calls are logged as tool spans. To log a retriever function as a retriever span (enabling RAG quality evaluators in Splunk Agent Observability), decorate it with @splunk_ao_retriever:

from splunk_ao_adk import splunk_ao_retriever
from google.adk.tools import FunctionTool

@splunk_ao_retriever
def search_docs(query: str) -> str:
    """Search the knowledge base."""
    # Your retrieval logic here
    results = my_vector_db.search(query)
    return "\n".join(r["content"] for r in results)

# Wrap in a FunctionTool - Splunk Agent Observability will log this as a retriever span
tool = FunctionTool(search_docs)

This enables RAG-specific evaluators like chunk attribution and context relevance to be calculated for your retrieval operations.

Custom ingestion hook

For advanced use cases, you can intercept traces for custom processing before forwarding to Splunk Agent Observability. This is useful for:

Custom session management
Local trace inspection and debugging
Filtering or enriching trace data

import asyncio
import os
from splunk_ao import SplunkAOLogger
from splunk_ao_adk import SplunkAOADKPlugin
from google.adk.runners import Runner
from google.adk.agents import LlmAgent
from google.genai import types

# Create a Splunk Agent Observability logger for custom trace handling
logger = SplunkAOLogger(
    project=os.getenv("SPLUNK_AO_PROJECT", "my-project"),
    agent_stream=os.getenv("SPLUNK_AO_AGENT_STREAM", "dev"),
)

def my_ingestion_hook(request):
    """Hook that captures traces locally and forwards to Splunk Agent Observability."""
    if hasattr(request, "traces") and request.traces:
        print(f"\n[Ingestion Hook] Intercepted {len(request.traces)} trace(s)")
        for trace in request.traces:
            spans = getattr(trace, "spans", []) or []
            span_types = [getattr(s, "type", "unknown") for s in spans]
            print(f"  - Trace with {len(spans)} span(s): {span_types}")

    # Session management: same external_id returns the same Splunk Agent Observability session
    splunk_ao_session_id = logger.start_session(external_id=request.session_external_id)
    request.session_id = splunk_ao_session_id

    # Forward traces to Splunk Agent Observability
    logger.ingest_traces(request)

async def main():
    # Create plugin with custom ingestion hook
    plugin = SplunkAOADKPlugin(ingestion_hook=my_ingestion_hook)

    agent = LlmAgent(
        name="assistant",
        model="gemini-2.0-flash",
        instruction="You are helpful."
    )
    runner = Runner(agent=agent, plugins=[plugin])

    message = types.Content(parts=[types.Part(text="Hello!")])
    async for event in runner.run_async(
        user_id="user-123",
        session_id="session-456",
        new_message=message
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)

if __name__ == "__main__":
    # Set environment variables: GOOGLE_API_KEY and
    # SPLUNK_AO_API_KEY (only for on-premises, standalone, and custom deployments)
    asyncio.run(main())

Plugin vs Callback mode

Feature	SplunkAOADKPlugin	SplunkAOADKCallback
Invocation tracking	Yes	No
Multi-agent support	Full	Limited
Sub-invocation detection	Yes	No
Setup complexity	Lower	Higher
Granular control	Lower	Higher

Recommendation: Use SplunkAOADKPlugin for most applications. Use SplunkAOADKCallback only when you need fine-grained control over individual agent callbacks.

Next steps

Splunk Agent Observability logger

Log with full control over sessions, traces, and spans using the Splunk Agent Observability logger.

OpenTelemetry integration

Use OpenTelemetry for Google ADK observability.

RAG evaluators

Learn about RAG quality evaluators for retriever spans.

Sessions

Learn how sessions group related traces together.

Overview

Get Started

Observability

Evaluators

Signals

AI Assistant

Luna Studio

Experiments

Agent Control

Annotations

Integrations

Security

References

Installation

Quick start

Configuration

Features

Session tracking

Custom metadata

Callback mode

Retriever spans

Custom ingestion hook

Plugin vs Callback mode

Next steps

Splunk Agent Observability logger

OpenTelemetry integration

RAG evaluators

Sessions

​Installation

​Quick start

​Configuration

​Features

​Session tracking

​Custom metadata

​Callback mode

​Retriever spans

​Custom ingestion hook

​Plugin vs Callback mode

​Next steps

Splunk Agent Observability logger

OpenTelemetry integration

RAG evaluators

Sessions

Installation

Quick start

Configuration

Features

Session tracking

Custom metadata

Callback mode

Retriever spans

Custom ingestion hook

Plugin vs Callback mode

Next steps