Tracing Google ADK Agents: Observability for Gemini-Powered Agent Pipelines

What Google ADK adds

Google's Agent Development Kit (ADK), released in 2025, provides a Python framework for building multi-agent applications powered by Gemini. It gives you three key agent types: Agent (single-step or multi-turn LLM agent with tools), SequentialAgent (chain of agents executed in order), and LoopAgent (agent that runs until a termination condition is met).

ADK's multi-agent model introduces failure modes common to all orchestration frameworks:

• LoopAgent termination failures: the loop condition never evaluates to true, causing the agent to consume tokens indefinitely until a hard limit is reached.
• Sequential step failures: a SequentialAgent step fails midway and the framework handles the error internally, making it hard to identify which step broke the chain.
• Tool call failures: ADK's FunctionTool wraps Python functions, but when the underlying function fails, the error is surfaced as an agent observation rather than a hard exception.
• Sub-agent output mismatch: when a parent agent passes a task to a sub-agent, the sub-agent's output format may not match what the parent expects, causing silent downstream failures.

Instrumenting a basic ADK Agent

ADK agents are run with runner.run_async() or runner.run(). Wrap the runner call in a Nexus trace:

import os
import time
import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from nexus_sdk import NexusClient

nexus = NexusClient(api_key=os.environ["NEXUS_API_KEY"])

# Define tools
def search_docs(query: str) -> str:
    """Search internal documentation."""
    return f"Documentation result for: {query}"

# Create agent
research_agent = Agent(
    name="research-agent",
    model="gemini-2.0-flash",
    instruction="You are a research assistant. Use search_docs to answer questions.",
    tools=[search_docs],
)

async def run_with_tracing(task: str, user_id: str) -> str:
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="research-app",
        user_id=user_id,
    )

    runner = Runner(
        agent=research_agent,
        app_name="research-app",
        session_service=session_service,
    )

    trace = nexus.start_trace({
        "agent_id": "google-adk-research-agent",
        "name": f"research: {task[:60]}",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {
            "user_id": user_id,
            "model": "gemini-2.0-flash",
            "environment": os.environ.get("APP_ENV", "dev"),
        },
    })
    trace_id = trace["trace_id"]

    t0 = time.time()
    try:
        from google.genai.types import Part, UserContent
        content = UserContent(parts=[Part(text=task)])

        final_response = None
        async for event in runner.run_async(
            user_id=user_id,
            session_id=session.id,
            new_message=content,
        ):
            if event.is_final_response():
                final_response = event.content.parts[0].text if event.content else ""

        nexus.end_trace(trace_id, {
            "status": "success",
            "latency_ms": int((time.time() - t0) * 1000),
            "metadata": {
                "response_length": len(final_response or ""),
            },
        })
        return final_response or ""

    except Exception as e:
        nexus.end_trace(trace_id, {
            "status": "error",
            "latency_ms": int((time.time() - t0) * 1000),
            "error": str(e),
        })
        raise

Tracing a SequentialAgent

SequentialAgents chain multiple agents in order. Emit a span for each step to identify which step in the chain is slow or failing:

from google.adk.agents import SequentialAgent, Agent

researcher = Agent(
    name="researcher",
    model="gemini-2.0-flash",
    instruction="Research the topic and return findings.",
    tools=[search_docs],
)

summarizer = Agent(
    name="summarizer",
    model="gemini-2.0-flash",
    instruction="Summarize the research findings concisely.",
)

pipeline = SequentialAgent(
    name="research-pipeline",
    sub_agents=[researcher, summarizer],
)

async def run_pipeline_with_tracing(task: str, user_id: str) -> str:
    trace = nexus.start_trace({
        "agent_id": "google-adk-research-pipeline",
        "name": f"pipeline: {task[:60]}",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {
            "user_id": user_id,
            "steps": ["researcher", "summarizer"],
        },
    })
    trace_id = trace["trace_id"]

    step_names = ["researcher", "summarizer"]
    step_index = [0]

    t0 = time.time()
    try:
        runner = Runner(
            agent=pipeline,
            app_name="research-pipeline",
            session_service=session_service,
        )

        final_response = None
        async for event in runner.run_async(
            user_id=user_id,
            session_id=session_id,
            new_message=content,
        ):
            # Track sub-agent transitions
            if hasattr(event, 'author') and event.author in step_names:
                current_step = event.author
                if event.is_final_response():
                    nexus.add_span(trace_id, {
                        "name": f"step:{current_step}",
                        "status": "success",
                        "latency_ms": int((time.time() - t0) * 1000),
                    })

            if event.is_final_response():
                final_response = event.content.parts[0].text if event.content else ""

        nexus.end_trace(trace_id, {
            "status": "success",
            "latency_ms": int((time.time() - t0) * 1000),
        })
        return final_response or ""

    except Exception as e:
        nexus.end_trace(trace_id, {
            "status": "error",
            "latency_ms": int((time.time() - t0) * 1000),
            "error": str(e),
        })
        raise

Tracing a LoopAgent

LoopAgents run until a condition is met or a maximum iteration count is reached. Track iteration count as a span to detect runaway loops:

from google.adk.agents import LoopAgent

refiner = Agent(
    name="refiner",
    model="gemini-2.0-flash",
    instruction="""Improve the draft. If quality score > 8, say DONE. Otherwise continue refining.""",
)

loop_agent = LoopAgent(
    name="refinement-loop",
    sub_agents=[refiner],
    max_iterations=5,
)

async def run_loop_with_tracing(draft: str, user_id: str) -> str:
    trace = nexus.start_trace({
        "agent_id": "google-adk-refinement-loop",
        "name": f"refine: {draft[:60]}",
        "status": "running",
        "started_at": nexus.now(),
        "metadata": {
            "user_id": user_id,
            "max_iterations": 5,
        },
    })
    trace_id = trace["trace_id"]

    iteration = 0
    t0 = time.time()
    try:
        final_output = None
        async for event in runner.run_async(...):
            if hasattr(event, 'author') and event.author == "refiner":
                if event.is_final_response():
                    iteration += 1
                    nexus.add_span(trace_id, {
                        "name": f"iteration:{iteration}",
                        "status": "success",
                        "latency_ms": int((time.time() - t0) * 1000),
                        "metadata": {"iteration": iteration},
                    })

            if event.is_final_response():
                final_output = event.content.parts[0].text if event.content else ""

        nexus.end_trace(trace_id, {
            "status": "success",
            "latency_ms": int((time.time() - t0) * 1000),
            "metadata": {"total_iterations": iteration},
        })
        return final_output or ""

    except Exception as e:
        nexus.end_trace(trace_id, {
            "status": "error",
            "latency_ms": int((time.time() - t0) * 1000),
            "error": str(e),
            "metadata": {"iterations_completed": iteration},
        })
        raise

What to track for Gemini tool calls

ADK's FunctionTool wraps Python functions and surfaces calls through the event stream. Wrap your tool functions to emit spans:

def traced_tool(tool_fn, tool_name: str, nexus_client, trace_id: str):
    def wrapper(*args, **kwargs):
        t0 = time.time()
        try:
            result = tool_fn(*args, **kwargs)
            nexus_client.add_span(trace_id, {
                "name": f"tool:{tool_name}",
                "status": "success",
                "latency_ms": int((time.time() - t0) * 1000),
                "metadata": {
                    "input": str(args[0])[:200] if args else str(kwargs)[:200],
                },
            })
            return result
        except Exception as e:
            nexus_client.add_span(trace_id, {
                "name": f"tool:{tool_name}",
                "status": "error",
                "latency_ms": int((time.time() - t0) * 1000),
                "error": str(e),
            })
            raise
    return wrapper

Next steps

Google ADK is newer than most agent frameworks — the ecosystem is growing fast and the observability tooling is still maturing. Nexus gives you agent-level health monitoring out of the box, regardless of framework. Sign up for a free Nexus account or read the Google ADK integration guide.