Observability for Google Agent Development Kit (Python) with Opik - Opik

<Note> In Opik 2.0, datasets and experiments are project-scoped. Make sure to specify a `project_name` when creating datasets and running experiments so they are associated with the correct project. </Note>

Agent Development Kit (ADK) is a flexible and modular framework for developing and deploying AI agents. ADK can be used with popular LLMs and open-source generative AI tools and is designed with a focus on tight integration with the Google ecosystem and Gemini models. ADK makes it easy to get started with simple agents powered by Gemini models and Google AI tools while providing the control and structure needed for more complex agent architectures and orchestration.

In this guide, we will showcase how to integrate Opik with Google ADK so that all the ADK calls are logged as traces in Opik. We'll cover three key integration patterns:

Automatic Agent Tracking - Recommended approach using track_adk_agent_recursive for effortless instrumentation
Manual Callback Configuration - Alternative approach with explicit callback setup for fine-grained control
Hybrid Tracing - Combining Opik decorators with ADK callbacks for comprehensive observability

Account Setup

Comet provides a hosted version of the Opik platform, simply create an account and grab your API Key.

You can also run the Opik platform locally, see the installation guide for more information.

Opik provides comprehensive integration with ADK, automatically logging traces for all agent executions, tool calls, and LLM interactions with detailed cost tracking and error monitoring.

Key Features

One-line instrumentation with track_adk_agent_recursive for automatic tracing of entire agent hierarchies
Automatic cost tracking for all supported LLM providers including LiteLLM models (OpenAI, Anthropic, Google AI, AWS Bedrock, and more)
Full compatibility with the @opik.track decorator for hybrid tracing approaches
Thread support for conversational applications using ADK sessions
Automatic agent graph visualization with Mermaid diagrams for complex multi-agent workflows
Comprehensive error tracking with detailed error information and stack traces

Getting Started

Installation

First, ensure you have both opik and google-adk installed:

bash

pip install opik google-adk

Configuring Opik

Configure the Opik Python SDK for your deployment type. See the Python SDK Configuration guide for detailed instructions on:

CLI configuration: opik configure
Code configuration: opik.configure()
Self-hosted vs Cloud vs Enterprise setup
Configuration files and environment variables

Configuring Google ADK

In order to configure Google ADK, you will need to have your LLM provider API key. For this example, we'll use OpenAI. You can find or create your OpenAI API Key in this page.

You can set it as an environment variable:

bash

export OPENAI_API_KEY="YOUR_API_KEY"

Or set it programmatically:

python

import os
import getpass

if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

Example 1: Automatic Agent Tracking (Recommended)

The recommended way to track ADK agents is using track_adk_agent_recursive and OpikTracer, which automatically instruments your entire agent hierarchy with a single function call. This approach is ideal for both single agents and complex multi-agent setups:

python

import datetime
from zoneinfo import ZoneInfo

from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from opik.integrations.adk import OpikTracer, track_adk_agent_recursive

def get_weather(city: str) -> dict:
    """Get weather information for a city."""
    if city.lower() == "new york":
        return {
            "status": "success",
            "report": "The weather in New York is sunny with a temperature of 25 °C (77 °F).",
        }
    elif city.lower() == "london":
        return {
            "status": "success",
            "report": "The weather in London is cloudy with a temperature of 18 °C (64 °F).",
        }
    return {"status": "error", "error_message": f"Weather info for '{city}' is unavailable."}

def get_current_time(city: str) -> dict:
    """Get current time for a city."""
    if city.lower() == "new york":
        tz = ZoneInfo("America/New_York")
        now = datetime.datetime.now(tz)
        return {
            "status": "success",
            "report": now.strftime(f"The current time in {city} is %Y-%m-%d %H:%M:%S %Z%z."),
        }
    elif city.lower() == "london":
        tz = ZoneInfo("Europe/London")
        now = datetime.datetime.now(tz)
        return {
            "status": "success",
            "report": now.strftime(f"The current time in {city} is %Y-%m-%d %H:%M:%S %Z%z."),
        }
    return {"status": "error", "error_message": f"No timezone info for '{city}'."}

# Initialize LiteLLM with OpenAI gpt-4o
llm = LiteLlm(model="openai/gpt-4o")

# Create the basic agent
basic_agent = LlmAgent(
    name="weather_time_agent",
    model=llm,
    description="Agent for answering time & weather questions",
    instruction="Answer questions about the time or weather in a city. Be helpful and provide clear information.",
    tools=[get_weather, get_current_time],
)

# Configure Opik tracer
opik_tracer = OpikTracer(
    name="basic-weather-agent",
    tags=["basic", "weather", "time", "single-agent"],
    metadata={
        "environment": "development",
        "model": "gpt-4o",
        "framework": "google-adk",
        "example": "basic"
    },
    project_name="adk-basic-demo"
)

# Instrument the agent with a single function call - this is the recommended approach
track_adk_agent_recursive(basic_agent, opik_tracer)

Each agent execution will now be automatically logged to the Opik platform with detailed trace information:

This approach automatically handles:

All agent callbacks (before/after agent, model, and tool executions)
Sub-agents and nested agent hierarchies
Agent tools that contain other agents
Complex workflows with minimal code

Example 2: Manual Callback Configuration (Alternative Approach)

For a fine-grained control over which callbacks to instrument, you can manually configure the OpikTracer callbacks. This approach gives you explicit control but requires more setup code:

python

# Configure Opik tracer (same as before)
opik_tracer = OpikTracer(
    name="basic-weather-agent",
    tags=["basic", "weather", "time", "single-agent"],
    metadata={
        "environment": "development",
        "model": "gpt-4o",
        "framework": "google-adk",
        "example": "basic"
    },
    project_name="adk-basic-demo"
)

# Create the agent with explicit callback configuration
basic_agent = LlmAgent(
    name="weather_time_agent",
    model=llm,
    description="Agent for answering time & weather questions",
    instruction="Answer questions about the time or weather in a city. Be helpful and provide clear information.",
    tools=[get_weather, get_current_time],
    before_agent_callback=opik_tracer.before_agent_callback,
    after_agent_callback=opik_tracer.after_agent_callback,
    before_model_callback=opik_tracer.before_model_callback,
    after_model_callback=opik_tracer.after_model_callback,
    before_tool_callback=opik_tracer.before_tool_callback,
    after_tool_callback=opik_tracer.after_tool_callback,
)

<Note> For most use cases, we recommend using `track_adk_agent_recursive` (shown in Example 1) as it requires less code and automatically handles complex agent hierarchies. </Note>

Example 3: Multi-Agent Setup with Hierarchical Tracing

This example demonstrates a complex multi-agent setup where we have specialized agents for different tasks. Using track_adk_agent_recursive, you can instrument the entire hierarchy with a single function call:

python

def get_detailed_weather(city: str) -> dict:
    """Get detailed weather information including forecast."""
    weather_data = {
        "new york": {
            "current": "Sunny, 25°C (77°F)",
            "humidity": "65%",
            "wind": "10 km/h NW",
            "forecast": "Partly cloudy tomorrow, high of 27°C"
        },
        "london": {
            "current": "Cloudy, 18°C (64°F)",
            "humidity": "78%",
            "wind": "15 km/h SW",
            "forecast": "Light rain expected tomorrow, high of 16°C"
        },
        "tokyo": {
            "current": "Partly cloudy, 22°C (72°F)",
            "humidity": "70%",
            "wind": "8 km/h E",
            "forecast": "Sunny tomorrow, high of 25°C"
        }
    }

    city_lower = city.lower()
    if city_lower in weather_data:
        data = weather_data[city_lower]
        return {
            "status": "success",
            "report": f"Weather in {city}: {data['current']}. Humidity: {data['humidity']}, Wind: {data['wind']}. {data['forecast']}"
        }
    return {"status": "error", "error_message": f"Detailed weather for '{city}' is unavailable."}

def get_world_time(city: str) -> dict:
    """Get time information for major world cities."""
    timezones = {
        "new york": "America/New_York",
        "london": "Europe/London",
        "tokyo": "Asia/Tokyo",
        "sydney": "Australia/Sydney",
        "paris": "Europe/Paris"
    }

    city_lower = city.lower()
    if city_lower in timezones:
        tz = ZoneInfo(timezones[city_lower])
        now = datetime.datetime.now(tz)
        return {
            "status": "success",
            "report": now.strftime(f"Current time in {city}: %A, %B %d, %Y at %I:%M %p %Z")
        }
    return {"status": "error", "error_message": f"Time zone info for '{city}' is unavailable."}

def get_travel_info(from_city: str, to_city: str) -> dict:
    """Get basic travel information between cities."""
    travel_data = {
        ("new york", "london"): {"flight_time": "7 hours", "time_diff": "+5 hours"},
        ("london", "new york"): {"flight_time": "8 hours", "time_diff": "-5 hours"},
        ("new york", "tokyo"): {"flight_time": "14 hours", "time_diff": "+14 hours"},
        ("tokyo", "new york"): {"flight_time": "13 hours", "time_diff": "-14 hours"},
        ("london", "tokyo"): {"flight_time": "12 hours", "time_diff": "+9 hours"},
        ("tokyo", "london"): {"flight_time": "11 hours", "time_diff": "-9 hours"},
    }

    route = (from_city.lower(), to_city.lower())
    if route in travel_data:
        data = travel_data[route]
        return {
            "status": "success",
            "report": f"Travel from {from_city} to {to_city}: Approximately {data['flight_time']} flight time. Time difference: {data['time_diff']}"
        }
    return {"status": "error", "error_message": f"Travel info for '{from_city}' to '{to_city}' is unavailable."}

# Weather specialist agent (no Opik callbacks needed)
weather_agent = LlmAgent(
    name="weather_specialist",
    model=llm,
    description="Specialized agent for detailed weather information",
    instruction="Provide comprehensive weather information including current conditions and forecasts. Be detailed and informative.",
    tools=[get_detailed_weather]
)

# Time specialist agent (no Opik callbacks needed)
time_agent = LlmAgent(
    name="time_specialist",
    model=llm,
    description="Specialized agent for world time information",
    instruction="Provide accurate time information for cities around the world. Include day of week and full date.",
    tools=[get_world_time]
)

# Travel specialist agent (no Opik callbacks needed)
travel_agent = LlmAgent(
    name="travel_specialist",
    model=llm,
    description="Specialized agent for travel information",
    instruction="Provide helpful travel information including flight times and time zone differences.",
    tools=[get_travel_info]
)

# Configure Opik tracer for multi-agent example
multi_agent_tracer = OpikTracer(
    name="multi-agent-coordinator",
    tags=["multi-agent", "coordinator", "weather", "time", "travel"],
    metadata={
        "environment": "development",
        "model": "gpt-4o",
        "framework": "google-adk",
        "example": "multi-agent",
        "agent_count": 4
    },
    project_name="adk-multi-agent-demo"
)

# Coordinator agent with sub-agents
coordinator_agent = LlmAgent(
    name="travel_coordinator",
    model=llm,
    description="Coordinator agent that delegates to specialized agents for weather, time, and travel information",
    instruction="""You are a travel coordinator that helps users with weather, time, and travel information.

    You have access to three specialized agents:
    - weather_specialist: For detailed weather information
    - time_specialist: For world time information
    - travel_specialist: For travel planning information

    Delegate appropriate queries to the right specialist agents and compile comprehensive responses for the user.""",
    tools=[],  # No direct tools, delegates to sub-agents
    sub_agents=[weather_agent, time_agent, travel_agent],
)

# Use track_adk_agent_recursive to instrument all agents at once
# This automatically adds callbacks to the coordinator and ALL sub-agents
from opik.integrations.adk import track_adk_agent_recursive
track_adk_agent_recursive(coordinator_agent, multi_agent_tracer)

The trace can now be viewed in the UI, showing the complete hierarchy:

The track_adk_agent_recursive approach is particularly powerful for:

Multi-agent systems with coordinator and specialist agents
Sequential agents with multiple processing steps
Parallel agents executing tasks concurrently
Loop agents with iterative workflows
Agent tools that contain nested agents
Complex hierarchies with deeply nested agent structures

By calling track_adk_agent_recursive once on the top-level agent, all child agents and their operations are automatically instrumented without any additional code

Cost Tracking

Opik automatically tracks token usage and cost for all LLM calls during the agent execution, not only for the Gemini LLMs, but including the models accessed via LiteLLM.

<Tip> View the complete list of supported models and providers on the [Supported Models](/tracing/supported_models) page. </Tip>

Agent Graph Visualization

Opik automatically generates visual representations of your agent workflows using Mermaid diagrams. The graph shows:

Agent hierarchy and relationships
Sequential execution flows
Parallel processing branches
Loop structures and iterations
Tool connections and dependencies

The graph is automatically computed and stored with each trace, providing a clear visual understanding of your agent's execution flow:

For weather time agent the graph will look like that:

For more complex agent architectures displaying a graph may be even more beneficial:

Example 4: Hybrid Tracing - Combining Opik Decorators with ADK Callbacks

This advanced example shows how to combine Opik's @opik.track decorator with ADK's callback system. This is powerful when you have complex multi-step tools that perform their own internal operations that you want to trace separately, while still maintaining the overall agent trace context.

You can use track_adk_agent_recursive together with @opik.track decorators on your tool functions for maximum visibility:

python

from opik import track

@track(name="weather_data_processing", tags=["data-processing", "weather"])
def process_weather_data(raw_data: dict) -> dict:
    """Process raw weather data with additional computations."""
    # Simulate some data processing steps that we want to trace separately
    processed = {
        "temperature_celsius": raw_data.get("temp_c", 0),
        "temperature_fahrenheit": raw_data.get("temp_c", 0) * 9/5 + 32,
        "conditions": raw_data.get("condition", "unknown"),
        "comfort_index": "comfortable" if 18 <= raw_data.get("temp_c", 0) <= 25 else "less comfortable"
    }
    return processed

@track(name="location_validation", tags=["validation", "location"])
def validate_location(city: str) -> dict:
    """Validate and normalize city names."""
    # Simulate location validation logic that we want to trace
    normalized_cities = {
        "nyc": "New York",
        "ny": "New York",
        "new york city": "New York",
        "london uk": "London",
        "london england": "London",
        "tokyo japan": "Tokyo"
    }

    city_lower = city.lower().strip()
    validated_city = normalized_cities.get(city_lower, city.title())

    return {
        "original": city,
        "validated": validated_city,
        "is_valid": city_lower in ["new york", "london", "tokyo"] or city_lower in normalized_cities
    }

@track(name="advanced_weather_lookup", tags=["weather", "api-simulation"])
def get_advanced_weather(city: str) -> dict:
    """Get weather with internal processing steps tracked by Opik decorators."""

    # Step 1: Validate location (traced by @opik.track)
    location_result = validate_location(city)

    if not location_result["is_valid"]:
        return {
            "status": "error",
            "error_message": f"Invalid location: {city}"
        }

    validated_city = location_result["validated"]

    # Step 2: Get raw weather data (simulated)
    raw_weather_data = {
        "New York": {"temp_c": 25, "condition": "sunny", "humidity": 65},
        "London": {"temp_c": 18, "condition": "cloudy", "humidity": 78},
        "Tokyo": {"temp_c": 22, "condition": "partly cloudy", "humidity": 70}
    }

    if validated_city not in raw_weather_data:
        return {
            "status": "error",
            "error_message": f"Weather data unavailable for {validated_city}"
        }

    raw_data = raw_weather_data[validated_city]

    # Step 3: Process the data (traced by @opik.track)
    processed_data = process_weather_data(raw_data)

    return {
        "status": "success",
        "city": validated_city,
        "report": f"Weather in {validated_city}: {processed_data['conditions']}, {processed_data['temperature_celsius']}°C ({processed_data['temperature_fahrenheit']:.1f}°F). Comfort level: {processed_data['comfort_index']}.",
        "raw_humidity": raw_data["humidity"]
    }

# Configure Opik tracer for hybrid example
hybrid_tracer = OpikTracer(
    name="hybrid-tracing-agent",
    tags=["hybrid", "decorators", "callbacks", "advanced"],
    metadata={
        "environment": "development",
        "model": "gpt-4o",
        "framework": "google-adk",
        "example": "hybrid-tracing",
        "tracing_methods": ["decorators", "callbacks"]
    },
    project_name="adk-hybrid-demo"
)

# Create hybrid agent that combines both tracing approaches
hybrid_agent = LlmAgent(
    name="advanced_weather_time_agent",
    model=llm,
    description="Advanced agent with hybrid Opik tracing using both decorators and callbacks",
    instruction="""You are an advanced weather and time agent that provides detailed information with comprehensive internal processing.

    Your tools perform multi-step operations that are individually traced, giving detailed visibility into the processing pipeline.
    Use the advanced weather and time tools to provide thorough, well-processed information to users.""",
    tools=[get_advanced_weather],
)

# Instrument the agent with track_adk_agent_recursive
# The @opik.track decorators in your tools will automatically create child spans
from opik.integrations.adk import track_adk_agent_recursive
track_adk_agent_recursive(hybrid_agent, hybrid_tracer)

The trace can now be viewed in the UI:

Compatibility with @track Decorator

The OpikTracer is fully compatible with the @track decorator, allowing you to create hybrid tracing approaches that combine ADK agent tracking with custom function tracing. You can both invoke your agent from inside another tracked function and call tracked functions inside your tool functions, all the spans and traces parent-child relationships will be preserved!

Thread Support

The Opik integration automatically handles ADK sessions and maps them to Opik threads for conversational applications:

python

from opik.integrations.adk import OpikTracer
from google.adk import sessions as adk_sessions, runners as adk_runners

# ADK session management
session_service = adk_sessions.InMemorySessionService()
session = session_service.create_session_sync(
    app_name="my_app",
    user_id="user_123",
    session_id="conversation_456"
)

opik_tracer = OpikTracer()
runner = adk_runners.Runner(
    agent=your_agent,
    app_name="my_app",
    session_service=session_service
)

# All traces will be automatically grouped by session_id as thread_id

The integration automatically:

Uses the ADK session ID as the Opik thread ID
Groups related conversations and interactions
Logs app_name and user_id as metadata
Maintains conversation context across multiple interactions

You can view your session as a whole conversation and easily navigate to any specific trace you need.

Error Tracking

The OpikTracer provides comprehensive error tracking and monitoring:

Automatic error capture for agent execution failures
Detailed stack traces with full context information
Tool execution errors with input/output data
Model call failures with provider-specific error details

Error information is automatically logged to spans and traces, making it easy to debug issues in production:

Troubleshooting: Missing Trace

When using Runner.run_async, make sure to process all events completely, even after finding the final response (when event.is_final_response() is True). If you exit the loop too early, OpikTracer won't log the final response and your trace will be incomplete. Don't use code that stops processing events prematurely:

python

async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
    if event.is_final_response():
        ...
        break  # Stop processing events once the final response is found

There is an upstream discussion about how to best solve this source of confusion: https://github.com/google/adk-python/issues/1695.

<Tip> Our team tried to address those issues and make the integration as robust as possible. If you are facing similar problems, the first thing we recommend is to update both `opik` and `google-adk` to the latest versions. We are actively working on improving this integration, so with the most recent versions you'll most likely get the best UX!. </Tip>

Flushing Traces

The OpikTracer object has a flush method that ensures all traces are logged to the Opik platform before you exit a script:

python

from opik.integrations.adk import OpikTracer

opik_tracer = OpikTracer()

# Your ADK agent execution code here...

# Ensure all traces are sent before script exits
opik_tracer.flush()

Prompts integration

The OpikTracer can be used together with the OPIK prompts library to easily access your existing prompts or create new ones, and then associate them with traces or spans within an ADK agent flow.

python

import datetime
import uuid
from typing import Iterator, Optional
from zoneinfo import ZoneInfo

from google.adk import Agent, Runner
from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from google.adk.sessions import InMemorySessionService
from google.genai import types as genai_types
from google.adk import events as adk_events

import opik
from opik.opik_context import update_current_trace
from opik.integrations.adk import OpikTracer, track_adk_agent_recursive

# Create prompt
system_prompt = opik.Prompt(
    name="system-prompt",
    prompt="You are a helpful assistant that provides accurate and concise answers.",
    project_name="my-project",
)

# Get prompt from the Prompt library
client = opik.Opik()
user_prompt = client.get_prompt(name="user-prompt")

def get_weather(city: str) -> dict:
    """Get weather information for a city."""
    # Add prompts to the current trace
    update_current_trace(
        prompts=[system_prompt, user_prompt]
    )

    if city.lower() == "new york":
        return {
            "status": "success",
            "report": "The weather in New York is sunny with a temperature of 25 °C (77 °F).",
        }
    elif city.lower() == "london":
        return {
            "status": "success",
            "report": "The weather in London is cloudy with a temperature of 18 °C (64 °F).",
        }
    return {"status": "error", "error_message": f"Weather info for '{city}' is unavailable."}

def get_current_time(city: str) -> dict:
    """Get current time for a city."""
    if city.lower() == "new york":
        tz = ZoneInfo("America/New_York")
        now = datetime.datetime.now(tz)
        return {
            "status": "success",
            "report": now.strftime(f"The current time in {city} is %Y-%m-%d %H:%M:%S %Z%z."),
        }
    elif city.lower() == "london":
        tz = ZoneInfo("Europe/London")
        now = datetime.datetime.now(tz)
        return {
            "status": "success",
            "report": now.strftime(f"The current time in {city} is %Y-%m-%d %H:%M:%S %Z%z."),
        }
    return {"status": "error", "error_message": f"No timezone info for '{city}'."}

# Initialize LiteLLM with OpenAI gpt-4o
llm = LiteLlm(model="openai/gpt-4o")

# Create the basic agent
basic_agent = LlmAgent(
    name="weather_time_agent",
    model=llm,
    description="Agent for answering time & weather questions",
    instruction="Answer questions about the time or weather in a city. Be helpful and provide clear information.",
    tools=[get_weather, get_current_time],
)

# Configure Opik tracer
opik_tracer = OpikTracer(
    name="basic-weather-agent",
    tags=["basic", "weather", "time", "single-agent"],
    metadata={
        "environment": "development",
        "model": "gpt-4o",
        "framework": "google-adk",
        "example": "basic"
    },
    project_name="adk-basic-demo"
)

# Instrument the agent with a single function call - this is the recommended approach
track_adk_agent_recursive(basic_agent, opik_tracer)