Google ADK

pip install raindrop-google-adk google-adk

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

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.",
)

session_service = InMemorySessionService()
runner = Runner(app_name="weather_app", agent=agent, session_service=session_service)

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):
    if event.is_final_response():
        print(event.content.parts[0].text)

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
    project_id="support-prod",     # Optional: route events to a specific project (slug)
    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

from raindrop_google_adk import create_raindrop_google_adk

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

wrapped_runner = rd.wrap(runner)

for event in wrapped_runner.run(user_id="user123", session_id="s1", new_message=msg):
    print(event)

rd.shutdown()

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

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

rd = setup_google_adk(
    api_key="your-write-key",
    project_id="support-prod",
)

from raindrop_google_adk import RaindropGoogleADK

# One long-lived wrapper per project, created at startup and reused.
rd_support = RaindropGoogleADK(api_key="your-write-key", project_id="support-prod")
rd_billing = RaindropGoogleADK(api_key="your-write-key", project_id="billing-prod")

# Wrap each runner with the corresponding wrapper.
support_runner = rd_support.wrap(support_runner)   # -> support-prod
billing_runner = rd_billing.wrap(billing_runner)   # -> billing-prod

rd_support.flush()
rd_billing.flush()

from raindrop import Raindrop

client = Raindrop(api_key="your-write-key", project_id="support-prod")
rd = RaindropGoogleADK(client=client)

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

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

def get_stock_price(symbol: str) -> str:
    """Get the current stock price."""
    return "189.50"

agent = LlmAgent(
    name="stock_agent",
    tools=[get_stock_price],
    model="gemini-2.5-flash",
    instruction="Answer questions about stock prices.",
)

# Tool calls are tracked as individual spans and counted in event properties as
# google_adk.tool_calls_count and google_adk.tool_call_names

from google.adk.agents import LlmAgent, SequentialAgent, ParallelAgent

researcher = LlmAgent(name="researcher", model="gemini-2.5-flash",
    instruction="Research the topic.", tools=[search_tool])
writer = LlmAgent(name="writer", model="gemini-2.5-flash",
    instruction="Write a summary based on research.")

pipeline = SequentialAgent(name="research_pipeline", sub_agents=[researcher, writer])

runner = Runner(app_name="research_app", agent=pipeline, session_service=session_service)
# Runner invocations are automatically traced with input/output capture

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

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

from raindrop_google_adk import setup_google_adk, create_raindrop_google_adk

# Auto-patch all Runner instances
rd = setup_google_adk(api_key="your-write-key")

# Or create without auto-patching
rd = create_raindrop_google_adk(api_key="your-write-key")
wrapped = rd.wrap(runner)