raindrop-google-adk 0.0.3

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())

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