Google ADK

import asyncio
import traceroot
from traceroot import Integration
from google.adk import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

traceroot.initialize(integrations=[Integration.GOOGLE_ADK])


def get_weather(city: str) -> dict:
    """Get current weather for a city."""
    return {"status": "success", "city": city, "temp": 72, "condition": "sunny"}


# ADK resolves the Gemini API key from GOOGLE_API_KEY in the environment.
agent = Agent(
    name="assistant",
    model="gemini-3.8-flash",
    description="Helpful assistant with access to tools.",
    instruction="You are a helpful AI assistant. Use tools when needed.",
    tools=[get_weather],
)


async def main():
    runner = InMemoryRunner(agent=agent, app_name="traceroot-adk-demo")
    await runner.session_service.create_session(
        app_name="traceroot-adk-demo",
        user_id="demo-user",
        session_id="demo-session",
    )

    # The full run, including tool calls and Gemini calls, is traced
    async for event in runner.run_async(
        user_id="demo-user",
        session_id="demo-session",
        new_message=types.Content(
            role="user",
            parts=[types.Part(text="What's the weather in Tokyo?")],
        ),
    ):
        if event.is_final_response() and event.content and event.content.parts:
            print(event.content.parts[0].text)


if __name__ == "__main__":
    asyncio.run(main())
    traceroot.flush()

# my_agent/agent.py
import traceroot
from traceroot import Integration
from google.adk import Agent

traceroot.initialize(integrations=[Integration.GOOGLE_ADK])


def get_current_time(city: str) -> dict:
    """Returns the current time in a specified city."""
    return {"status": "success", "city": city, "time": "10:30 AM"}


root_agent = Agent(
    model="gemini-3.8-flash",
    name="root_agent",
    description="Tells the current time in a specified city.",
    instruction="Use the 'get_current_time' tool to answer.",
    tools=[get_current_time],
)

import asyncio
import traceroot
from traceroot import Integration
from google.adk import Agent, Event, Workflow
from google.adk.runners import InMemoryRunner
from google.adk.workflow import JoinNode
from google.genai import types

traceroot.initialize(integrations=[Integration.GOOGLE_ADK])


def record_city(node_input: str):
    """Store the city in session state and pass it on to both branches."""
    city = node_input.strip()
    return Event(output=city, state={"city": city})


def get_weather(city: str) -> dict:
    """Get current weather for the city in session state."""
    return {"city": city, "temp": 72, "condition": "sunny"}


suggest_food = Agent(
    name="suggest_food",
    model="gemini-3.8-flash",
    instruction="In one sentence, name a signature local dish to try in {city}.",
)

# START -> record_city -> (get_weather, suggest_food in parallel) -> gather
workflow = Workflow(
    name="city_brief",
    edges=[("START", record_city, (get_weather, suggest_food), JoinNode(name="gather"))],
)


async def main():
    runner = InMemoryRunner(agent=workflow, app_name="traceroot-adk-workflow")
    await runner.session_service.create_session(
        app_name="traceroot-adk-workflow",
        user_id="demo-user",
        session_id="demo-session",
    )

    # The workflow run and every node in it are traced
    output = None
    async for event in runner.run_async(
        user_id="demo-user",
        session_id="demo-session",
        new_message=types.Content(role="user", parts=[types.Part(text="Tokyo")]),
    ):
        if event.output is not None:
            output = event.output
    # The last output is the joined result, keyed by branch name
    print(output)


if __name__ == "__main__":
    asyncio.run(main())
    traceroot.flush()