raindrop-google-adk 0.0.8

Raindrop integration for Google ADK (Agent Development Kit)

raindrop-google-adk

Raindrop integration for Google ADK (Agent Development Kit) — Google's framework for building AI agents powered by Gemini models.

Wraps Google ADK Runner objects to automatically capture agent invocations, tool calls, and multi-step reasoning, shipping telemetry to Raindrop.

Installation

pip install raindrop-google-adk google-adk

Quick Start

from raindrop_google_adk import setup_google_adk
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Enable automatic tracing for all ADK Runner interactions
rd = setup_google_adk(api_key="your-write-key")

# Create your ADK agent as normal
def get_weather(city: str) -> dict:
    """Get weather for a city."""
    return {"temperature": 72, "condition": "sunny", "city": city}

agent = LlmAgent(
    name="weather_assistant",
    tools=[get_weather],
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant that can check weather.",
)

# Create session and runner
session_service = InMemorySessionService()
runner = Runner(app_name="weather_app", agent=agent, session_service=session_service)

# Use the runner as normal — all interactions are automatically traced
user_msg = types.Content(
    parts=[types.Part(text="What's the weather in New York?")],
    role="user",
)
for event in runner.run(user_id="user123", session_id="session123", new_message=user_msg):
    print(event)

What Gets Traced

The Google ADK integration automatically captures:

• Runner invocations — input message, user_id, session_id, app_name
• Agent responses — final output text from the agent
• Token usage — prompt_tokens, completion_tokens, total_tokens from usage metadata
• Tool calls — individual tool spans with name, input, output, duration, and error
• Model info — model version when available
• Agent identity — agent name, author from events
• Finish reason — why generation stopped (e.g., STOP, SAFETY, MAX_TOKENS)
• Errors — captured (with error message) and re-raised to the caller
• Async support — both run() (sync) and run_async() (async) are instrumented

Configuration

Auto-instrumentation (recommended)

from raindrop_google_adk import setup_google_adk

rd = setup_google_adk(
    api_key="your-write-key",      # Required: your Raindrop API key. If None, telemetry is disabled.
    user_id="user-123",            # Optional: default user ID for all events
    convo_id="convo-456",          # Optional: conversation/thread ID
    tracing_enabled=True,          # Optional: enable/disable OTEL tracing (default: True)
    bypass_otel_for_tools=True,    # Optional: bypass OTEL for tool calls (default: True)
)

# All Runner.run() and Runner.run_async() calls are now automatically traced
# Call rd.shutdown() before process exit

Manual wrapping

from raindrop_google_adk import create_raindrop_google_adk

rd = create_raindrop_google_adk(
    api_key="your-write-key",
    user_id="user-123",
)

# Wrap a specific runner instance
wrapped_runner = rd.wrap(runner)

# Use wrapped_runner as normal
for event in wrapped_runner.run(...):
    print(event)

rd.shutdown()

Class-based API

from raindrop_google_adk import RaindropGoogleADK

rd = RaindropGoogleADK(
    api_key="your-write-key",
    user_id="user-123",
    tracing_enabled=True,
    bypass_otel_for_tools=True,
    debug=False,
)

# Auto-patch all Runner instances
rd.setup()

# Or wrap a specific runner
wrapped = rd.wrap(runner)

# Cleanup
rd.shutdown()

Debug Mode

Enable verbose logging to troubleshoot integration issues:

rd = RaindropGoogleADK(
    api_key="your-write-key",
    debug=True,  # Enables DEBUG-level logging
)

Identify

Associate a user with traits for downstream analysis:

rd.identify(
    user_id="user-123",
    traits={"plan": "pro", "company": "Acme"},
)

Track Signal

Track feedback, edits, or custom signals tied to a specific event:

rd.track_signal(
    event_id="evt-abc123",
    name="thumbs_up",
    signal_type="feedback",
    sentiment="POSITIVE",
    comment="Great response!",
)

Async Usage

The wrapper supports both sync and async runner usage:

import asyncio

async def main():
    user_msg = types.Content(
        parts=[types.Part(text="What's the weather?")],
        role="user",
    )
    async for event in runner.run_async(
        user_id="user123",
        session_id="session123",
        new_message=user_msg,
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)

    rd.shutdown()

asyncio.run(main())

For multi-agent runs, one Raindrop ai_generation event is recorded per Runner.run() invocation. Since ADK can produce a final response for each participating agent, the event output uses the last non-empty final response; earlier agent responses are not merged into the conversation output. Model and agent metadata describe that selected response, while token and tool counts cover the full invocation.

Captured Properties

Each event includes the following properties when available:

Property	Description
ai.usage.prompt_tokens	Input token count
ai.usage.completion_tokens	Output token count
ai.usage.total_tokens	Total token count
google_adk.app_name	Runner app name
google_adk.user_id	User ID from the runner call
google_adk.session_id	Session ID from the runner call
google_adk.author	Event author (agent name)
google_adk.agent_name	Agent name from the event
google_adk.tool_calls_count	Number of tool calls in the run
google_adk.tool_call_names	JSON list of tool names called
google_adk.error	Whether a Python exception occurred
google_adk.error_message	Python exception message if applicable
google_adk.error_code	LLM-level error code (e.g. SAFETY, QUOTA)
google_adk.llm_error_message	LLM-level error message
google_adk.agent_branch	Agent hierarchy path for multi-agent setups
google_adk.finish_reason	Why generation stopped (STOP, SAFETY, etc.)
ai.usage.cached_tokens	Cached content token count
ai.usage.thoughts_tokens	Thinking/reasoning token count

Flushing and Shutdown

Always call shutdown() before your process exits to ensure all telemetry is shipped:

rd.flush()     # flush pending events without releasing resources
rd.shutdown()  # flush + release resources

Full Documentation

See the full documentation for detailed API reference, configuration options, and advanced usage.

Testing

cd packages/google-adk-python
pip install -e ".[dev]"
pytest