Agent Development Kit

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.

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

• 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
• Multi-agent instrumentation with track_adk_agent_recursive for complex agent hierarchies

Getting started

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

$
pip install opik google-adk

You can also set up Opik by using the opik configure CLI command. This will ask you to enter either your local server address or your API key if you’re using the Cloud version:

$
opik configure

Using OpikTracer

The OpikTracer automatically captures detailed information about your ADK agent executions, including inputs, outputs, metadata, token usage, and error information. Here’s a basic example:

1
import datetime
2
from zoneinfo import ZoneInfo
3
4
from google.adk.agents import Agent
5
from opik.integrations.adk import OpikTracer
6
7
8
def get_weather(city: str) -> dict:
9
    if city.lower() == "new york":
10
        return {
11
            "status": "success",
12
            "report": (
13
                "The weather in New York is sunny with a temperature of 25 degrees"
14
                " Celsius (41 degrees Fahrenheit)."
15
            ),
16
        }
17
    else:
18
        return {
19
            "status": "error",
20
            "error_message": f"Weather information for '{city}' is not available.",
21
        }
22
23
24
def get_current_time(city: str) -> dict:
25
    if city.lower() == "new york":
26
        tz_identifier = "America/New_York"
27
    else:
28
        return {
29
            "status": "error",
30
            "error_message": (f"Sorry, I don't have timezone information for {city}."),
31
        }
32
33
    tz = ZoneInfo(tz_identifier)
34
    now = datetime.datetime.now(tz)
35
    report = f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
36
    return {"status": "success", "report": report}
37
38
39
opik_tracer = OpikTracer()
40
41
root_agent = Agent(
42
    name="weather_time_agent",
43
    model="gemini-2.5-flash",
44
    description=("Agent to answer questions about the time and weather in a city."),
45
    instruction=("I can answer your questions about the time and weather in a city."),
46
    tools=[get_weather, get_current_time],
47
    before_agent_callback=opik_tracer.before_agent_callback,
48
    after_agent_callback=opik_tracer.after_agent_callback,
49
    before_model_callback=opik_tracer.before_model_callback,
50
    after_model_callback=opik_tracer.after_model_callback,
51
    before_tool_callback=opik_tracer.before_tool_callback,
52
    after_tool_callback=opik_tracer.after_tool_callback,
53
)

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

Multi-agent instrumentation with track_adk_agent_recursive

For complex multi-agent systems, manually adding OpikTracer callbacks to every agent can be time-consuming. The track_adk_agent_recursive function automatically adds OpikTracer callbacks to all agents in your hierarchy:

1
from opik.integrations.adk import OpikTracer, track_adk_agent_recursive
2
from google.adk import agents as adk_agents
3
4
opik_tracer = OpikTracer(project_name="multi-agent-system")
5
6
# Create a complex multi-agent system
7
translator_to_english = adk_agents.Agent(
8
    name="Translator",
9
    model="gemini-2.5-flash",
10
    description="Translates text to English.",
11
)
12
13
summarizer = adk_agents.Agent(
14
    name="Summarizer", 
15
    model="gemini-2.5-flash",
16
    description="Summarizes text to 1 sentence.",
17
)
18
19
root_agent = adk_agents.SequentialAgent(
20
    name="TextProcessingAssistant",
21
    sub_agents=[translator_to_english, summarizer],
22
    description="Runs translator to english then summarizer, in order.",
23
)
24
25
# Automatically instrument all agents in the hierarchy
26
track_adk_agent_recursive(root_agent, opik_tracer)
27
28
# Now all agents will be automatically tracked

This approach is particularly useful for:

• 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

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.

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:

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:

1
from opik.integrations.adk import OpikTracer
2
from google.adk import sessions as adk_sessions, runners as adk_runners
3
4
# ADK session management
5
session_service = adk_sessions.InMemorySessionService()
6
session = session_service.create_session_sync(
7
    app_name="my_app",
8
    user_id="user_123", 
9
    session_id="conversation_456"
10
)
11
12
opik_tracer = OpikTracer()
13
runner = adk_runners.Runner(
14
    agent=your_agent,
15
    app_name="my_app", 
16
    session_service=session_service
17
)
18
19
# 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:

1
async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
2
    if event.is_final_response():
3
        ...
4
        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.

Flushing Traces

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

1
from opik.integrations.adk import OpikTracer
2
3
opik_tracer = OpikTracer()
4
5
# Your ADK agent execution code here...
6
7
# Ensure all traces are sent before script exits
8
opik_tracer.flush()