Google ADK Integration

pip install mantisdk google-adk openinference-instrumentation-google-adk

from dotenv import load_dotenv
load_dotenv()
 
# Initialize tracing FIRST
import mantisdk.tracing as tracing
 
tracing.init(service_name="adk-agent", instrument_default=False)
tracing.instrument(names=["google_adk"])
 
# NOW import Google ADK
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types
 
# Define a tool
def get_weather(city: str) -> dict:
    return {"status": "success", "weather": f"Sunny in {city}"}
 
# Create agent
agent = Agent(
    name="weather_agent",
    model="gemini-2.0-flash",
    tools=[get_weather],
)
 
# Run agent - automatically traced
runner = InMemoryRunner(agent=agent, app_name="weather_app")
 
# ... use runner.run_async() to interact
 
tracing.shutdown()

import uuid
from google.adk.runners import InMemoryRunner
 
runner = InMemoryRunner(agent=agent, app_name="my_app")
session_service = runner.session_service
 
user_id = f"user_{uuid.uuid4().hex[:8]}"
session_id = f"session_{uuid.uuid4().hex[:8]}"
 
await session_service.create_session(
    app_name="my_app",
    user_id=user_id,
    session_id=session_id,
)
 
# Sessions are traced

#!/usr/bin/env python3
"""Google ADK Research Assistant with Mantis SDK Tracing."""
 
import datetime
from typing import Optional
from zoneinfo import ZoneInfo
from dotenv import load_dotenv
load_dotenv()
 
# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured
 
 
def setup_tracing(trace_to_console: bool = False):
    """Initialize mantisdk tracing with Google ADK instrumentation."""
    exporters = []
 
    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")
    else:
        print("[Tracing] No Insight credentials found")
 
    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())
 
    tracing.init(
        service_name="google-adk-research-assistant",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )
 
    tracing.instrument(names=["google_adk"])
    print("[Tracing] Google ADK instrumentation enabled")
 
 
# NOW import Google ADK
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types
 
 
# Define tools
def get_current_time(city: str) -> dict:
    """Get the current time in a city."""
    timezone_map = {
        "new york": "America/New_York",
        "london": "Europe/London",
        "tokyo": "Asia/Tokyo",
        "paris": "Europe/Paris",
    }
 
    tz_id = timezone_map.get(city.lower())
    if not tz_id:
        return {"status": "error", "error_message": f"Unknown city: {city}"}
 
    tz = ZoneInfo(tz_id)
    now = datetime.datetime.now(tz)
    return {"status": "success", "time": now.strftime("%Y-%m-%d %H:%M:%S %Z")}
 
 
def get_weather(city: str) -> dict:
    """Get weather for a city (simulated)."""
    weather_data = {
        "new york": {"temp": 22, "condition": "Partly cloudy"},
        "london": {"temp": 15, "condition": "Rainy"},
        "tokyo": {"temp": 28, "condition": "Sunny"},
        "paris": {"temp": 18, "condition": "Overcast"},
    }
 
    data = weather_data.get(city.lower())
    if not data:
        return {"status": "error", "error_message": f"No weather data for {city}"}
 
    return {
        "status": "success",
        "weather": f"{data['condition']}, {data['temp']}°C"
    }
 
 
def calculate(expression: str) -> dict:
    """Evaluate a math expression."""
    import math
    safe_dict = {
        "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
        "pi": math.pi, "e": math.e, "abs": abs, "round": round,
    }
    try:
        result = eval(expression.replace("^", "**"), {"__builtins__": {}}, safe_dict)
        return {"status": "success", "result": result}
    except Exception as e:
        return {"status": "error", "error_message": str(e)}
 
 
def search_knowledge(query: str, topic: Optional[str] = None) -> dict:
    """Search a knowledge base (simulated)."""
    knowledge = {
        "python": "Python is a high-level programming language.",
        "machine learning": "ML enables systems to learn from data.",
        "quantum computing": "Uses quantum mechanics for computation.",
    }
 
    for key, content in knowledge.items():
        if query.lower() in key:
            return {"status": "success", "result": content}
 
    return {"status": "not_found", "message": f"No results for '{query}'"}
 
 
# Create the agent
def create_research_assistant(model: str = "gemini-2.0-flash") -> Agent:
    """Create the research assistant agent."""
    return Agent(
        name="research_assistant",
        model=model,
        description="A research assistant with time, weather, calculator, and knowledge tools.",
        instruction="""You are a helpful research assistant with access to:
1. get_current_time - Get time in various cities
2. get_weather - Get weather information
3. calculate - Evaluate math expressions
4. search_knowledge - Search a knowledge base
 
Use these tools to help answer user questions.""",
        tools=[get_current_time, get_weather, calculate, search_knowledge],
    )
 
 
async def run_query(agent: Agent, query: str):
    """Run a single query against the agent."""
    import uuid
 
    runner = InMemoryRunner(agent=agent, app_name="research_app")
    session_service = runner.session_service
 
    user_id = f"user_{uuid.uuid4().hex[:8]}"
    session_id = f"session_{uuid.uuid4().hex[:8]}"
 
    await session_service.create_session(
        app_name="research_app",
        user_id=user_id,
        session_id=session_id,
    )
 
    message = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
 
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=message,
    ):
        if event.is_final_response() and event.content:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    return part.text
 
    return None
 
 
async def main():
    import argparse
 
    parser = argparse.ArgumentParser()
    parser.add_argument("--query", "-q", default="What time is it in Tokyo?")
    parser.add_argument("--trace-to-console", action="store_true")
    args = parser.parse_args()
 
    setup_tracing(trace_to_console=args.trace_to_console)
 
    agent = create_research_assistant()
 
    print(f"Query: {args.query}")
    response = await run_query(agent, args.query)
    print(f"Response: {response}")
 
    tracing.shutdown()
 
 
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

run_async (auto-instrumented)
├── session_id: session_abc123
├── user_id: user_xyz789
├── agent: research_assistant
└── tool_calls
    └── get_current_time
        ├── input: {"city": "Tokyo"}
        └── output: {"status": "success", "time": "..."}

import mantisdk.tracing as tracing
from google.adk.runners import InMemoryRunner
 
async def enhanced_query(agent, query: str, user_id: str):
    with tracing.trace("research_session", input=query) as span:
        span.set_attribute(tracing.semconv.USER_ID, user_id)
        span.set_attribute(tracing.semconv.SESSION_ID, "session-123")
 
        runner = InMemoryRunner(agent=agent, app_name="research_app")
 
        # ADK calls are auto-instrumented as children
        response = await run_agent(runner, query)
 
        span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
        return response

async def run_interactive_session(agent: Agent):
    """Interactive CLI with session tracking."""
    import uuid
 
    runner = InMemoryRunner(agent=agent, app_name="research_app")
    session_service = runner.session_service
 
    user_id = f"user_{uuid.uuid4().hex[:8]}"
    session_id = f"session_{uuid.uuid4().hex[:8]}"
 
    await session_service.create_session(
        app_name="research_app",
        user_id=user_id,
        session_id=session_id,
    )
 
    print("Research Assistant ready. Type 'quit' to exit.")
 
    while True:
        user_input = input("You: ").strip()
        if user_input.lower() in ("quit", "exit"):
            break
 
        message = types.Content(
            role="user",
            parts=[types.Part(text=user_input)]
        )
 
        async for event in runner.run_async(
            user_id=user_id,
            session_id=session_id,
            new_message=message,
        ):
            if event.is_final_response() and event.content:
                for part in event.content.parts:
                    if hasattr(part, "text") and part.text:
                        print(f"Assistant: {part.text}")

# Google API key (required)
GOOGLE_API_KEY=your-google-api-key
 
# Insight credentials
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-...
INSIGHT_SECRET_KEY=sk-lf-...