Google ADK

Google ADK integration captures traces from your Google Agent Development Kit applications, including agent execution flow, LLM interactions, and tool calls. Since ADK uses the global OpenTelemetry tracer provider, Judgment captures spans out of the box with zero additional instrumentation.

Quickstart

Install Dependencies

uv add google-adk judgeval python-dotenv

pip install google-adk judgeval python-dotenv

Initialize Integration

setup.py

from judgeval.trace import JudgmentTracerProvider, Tracer

Tracer.init(project_name="adk-demo")
JudgmentTracerProvider.install_as_global_tracer_provider()

Google ADK uses the global OpenTelemetry tracer provider internally. By installing JudgmentTracerProvider as the global provider, every span ADK creates is automatically routed to Judgment — no extra instrumentation library required.

Add to Existing Code

Add these lines to your existing Google ADK application:

import asyncio
from dotenv import load_dotenv

load_dotenv()

from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types
from judgeval.trace import JudgmentTracerProvider, Tracer  

Tracer.init(project_name="adk-demo")  
JudgmentTracerProvider.install_as_global_tracer_provider()  

def get_current_time(city: str) -> dict:
    """Returns the current time in a specified city."""
    return {"status": "success", "city": city, "time": "10:30 AM"}

root_agent = Agent(
    model="gemini-2.0-flash",
    name="time_agent",
    description="Tells the current time in a specified city.",
    instruction=(
        "You are a helpful assistant. Use the get_current_time tool "
        "when asked about the time in a city."
    ),
    tools=[get_current_time],
)

async def main():
    runner = InMemoryRunner(agent=root_agent, app_name="demo")
    session = await runner.session_service.create_session(
        app_name="demo", user_id="user1"
    )
    msg = types.Content(
        role="user", parts=[types.Part(text="What time is it in Tokyo?")]
    )
    async for event in runner.run_async(
        user_id="user1", session_id=session.id, new_message=msg
    ):
        if event.content and event.content.parts:
            for p in event.content.parts:
                if p.text:
                    print(p.text)

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

All agent executions and tool calls are automatically traced.

Example: Multi-Agent Delegation

multi_agent_example.py

import asyncio
from dotenv import load_dotenv

load_dotenv()

from judgeval.trace import JudgmentTracerProvider, Tracer
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

Tracer.init(project_name="adk-demo")
JudgmentTracerProvider.install_as_global_tracer_provider()


def get_current_time(city: str) -> dict:
    """Returns the current time in a specified city."""
    return {"status": "success", "city": city, "time": "10:30 AM"}


def get_weather(city: str) -> dict:
    """Returns the current weather in a specified city."""
    return {"status": "success", "city": city, "weather": "sunny", "temp": "72°F"}


# Define specialized agents
time_agent = Agent(
    model="gemini-2.0-flash",
    name="time_agent",
    description="Provides the current time for a given city.",
    instruction="Use the get_current_time tool to answer time-related questions.",
    tools=[get_current_time],
)

weather_agent = Agent(
    model="gemini-2.0-flash",
    name="weather_agent",
    description="Provides current weather information for a given city.",
    instruction="Use the get_weather tool to answer weather-related questions.",
    tools=[get_weather],
)

# Define a coordinator agent that delegates to sub-agents
coordinator = Agent(
    model="gemini-2.0-flash",
    name="coordinator",
    description="Routes user questions to the appropriate specialized agent.",
    instruction=(
        "You are a coordinator. Delegate time questions to the time_agent "
        "and weather questions to the weather_agent."
    ),
    sub_agents=[time_agent, weather_agent],
)


@Tracer.observe(span_type="function")  
async def main():
    runner = InMemoryRunner(agent=coordinator, app_name="demo")
    session = await runner.session_service.create_session(
        app_name="demo", user_id="user1"
    )

    questions = [
        "What time is it in Tokyo?",
        "What's the weather like in Paris?",
    ]

    for question in questions:
        msg = types.Content(
            role="user", parts=[types.Part(text=question)]
        )
        print(f"Question: {question}")
        async for event in runner.run_async(
            user_id="user1", session_id=session.id, new_message=msg
        ):
            if event.content and event.content.parts:
                for p in event.content.parts:
                    if p.text:
                        print(f"Response: {p.text}")
        print()


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

Tracking Additional Operations: Use @Tracer.observe() to track any function or method outside the ADK workflow. This is especially useful for monitoring utility functions, API calls, or other operations that are part of your overall application flow.

complete_example.py

import asyncio
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types
from judgeval.trace import JudgmentTracerProvider, Tracer

Tracer.init(project_name="adk-demo")
JudgmentTracerProvider.install_as_global_tracer_provider()

@Tracer.observe(span_type="function")
def preprocess_input(data: str) -> str:
    return f"Preprocessed: {data}"

async def run_agent(user_input: str):
    # Preprocess input (traced with @Tracer.observe)
    processed_input = preprocess_input(user_input)

    agent = Agent(
        model="gemini-2.0-flash",
        name="assistant",
        description="A helpful assistant.",
        instruction="You are a helpful assistant.",
    )

    runner = InMemoryRunner(agent=agent, app_name="demo")
    session = await runner.session_service.create_session(
        app_name="demo", user_id="user1"
    )
    msg = types.Content(
        role="user", parts=[types.Part(text=processed_input)]
    )
    async for event in runner.run_async(
        user_id="user1", session_id=session.id, new_message=msg
    ):
        if event.content and event.content.parts:
            for p in event.content.parts:
                if p.text:
                    print(p.text)

# Execute - both helper functions and agents are traced
asyncio.run(run_agent("Hello World"))

Next Steps

Agent Behavior Monitoring - Monitor your Google ADK agents in production with behavioral scoring.
Tracing Reference - Learn more about Judgment's tracing capabilities and advanced configuration.