Google ADK → OpenObserve

pip install openobserve-python-sdk openinference-instrumentation-google-adk google-adk python-dotenv

# OpenObserve instance URL
# Default for self-hosted: http://localhost:5080
OPENOBSERVE_URL=https://api.openobserve.ai/

# Your OpenObserve organisation slug or ID
OPENOBSERVE_ORG=your_org_id

# Basic auth token — Base64-encoded "email:password"
OPENOBSERVE_AUTH_TOKEN=Basic <your_base64_token>

# Google AI Studio API key
GOOGLE_API_KEY=your-google-api-key

from dotenv import load_dotenv
load_dotenv()

from openinference.instrumentation.google_adk import GoogleADKInstrumentor
from openobserve import openobserve_init

GoogleADKInstrumentor().instrument()
openobserve_init()

import asyncio
from google.adk import Agent, Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types


def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"Sunny, 22°C in {city}"


agent = Agent(
    name="weather_agent",
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant. Use get_weather for weather questions.",
    tools=[get_weather],
)

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


async def main():
    await session_service.create_session(
        app_name="my_app", user_id="user1", session_id="session1"
    )
    content = types.Content(
        role="user",
        parts=[types.Part.from_text(text="What's the weather in London?")],
    )
    async for event in runner.run_async(
        user_id="user1", session_id="session1", new_message=content
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if getattr(part, "text", None):
                    print(part.text)


asyncio.run(main())