Google ADK

pip install arize-otel openinference-instrumentation-google-adk google-adk

export ARIZE_SPACE_ID="<your-space-id>"
export ARIZE_API_KEY="<your-api-key>"
export ARIZE_PROJECT_NAME="google-adk-tracing-example"
export GEMINI_API_KEY="<your-gemini-api-key>"

# instrumentation.py
import os

from arize.otel import register
from openinference.instrumentation.google_adk import GoogleADKInstrumentor

tracer_provider = register(
    space_id=os.environ["ARIZE_SPACE_ID"],
    api_key=os.environ["ARIZE_API_KEY"],
    project_name=os.environ["ARIZE_PROJECT_NAME"],
)

GoogleADKInstrumentor().instrument(tracer_provider=tracer_provider)
print("Arize AX tracing initialized for Google ADK.")

# example.py

# Importing instrumentation first ensures tracing is set up
# before `google.adk` is imported.
from instrumentation import tracer_provider

import asyncio

from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types


def get_ocean_salinity() -> dict:
    """Return a one-line factoid about ocean salinity."""
    return {
        "status": "success",
        "report": (
            "The ocean is salty because rivers continuously dissolve "
            "mineral salts from rocks and soil and carry them to the "
            "sea, where they accumulate over millions of years."
        ),
    }


agent = Agent(
    name="ocean_agent",
    model="gemini-2.5-flash",
    description="Answers questions about the ocean using a tool.",
    instruction=(
        "Use get_ocean_salinity to answer the user. "
        "Reply with a concise two-sentence summary."
    ),
    tools=[get_ocean_salinity],
)


async def main() -> None:
    runner = InMemoryRunner(agent=agent, app_name="ocean_app")
    await runner.session_service.create_session(
        app_name="ocean_app",
        user_id="demo_user",
        session_id="demo_session",
    )
    async for event in runner.run_async(
        user_id="demo_user",
        session_id="demo_session",
        new_message=types.Content(
            role="user",
            parts=[types.Part(text="Why is the ocean salty?")],
        ),
    ):
        if event.is_final_response():
            print(event.content.parts[0].text.strip())


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