Multi-Tool Single Agent

import datetime
from zoneinfo import ZoneInfo


def get_weather(city: str) -> dict:
    """Retrieves the current weather report for a specified city.

    Args:
        city (str): The name of the city for which to retrieve the weather report.

    Returns:
        dict: status and result or error msg.
    """
    if city.lower() == "new york":
        return {
            "status": "success",
            "report": (
                "The weather in New York is sunny with a temperature of 25 degrees"
                " Celsius (77 degrees Fahrenheit)."
            ),
        }
    else:
        return {
            "status": "error",
            "error_message": f"Weather information for '{city}' is not available.",
        }


def get_current_time(city: str) -> dict:
    """Returns the current time in a specified city.

    Args:
        city (str): The name of the city for which to retrieve the current time.

    Returns:
        dict: status and result or error msg.
    """

    if city.lower() == "new york":
        tz_identifier = "America/New_York"
    else:
        return {
            "status": "error",
            "error_message": (
                f"Sorry, I don't have timezone information for {city}."
            ),
        }

    tz = ZoneInfo(tz_identifier)
    now = datetime.datetime.now(tz)
    report = (
        f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
    )
    return {"status": "success", "report": report}

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
import os


root_agent = Agent(
    name="weather_time_agent",
    model=LiteLlm(
        model="openrouter/meta-llama/llama-4-scout:free",
        api_key=os.getenv("OpenRouter_API"),
        api_base=os.getenv("OpenRouter_URL"),
    ),
    description=(
        "Agent to answer questions about the time and weather in a city."
    ),
    instruction=(
        "You are a helpful agent who can answer user questions \
        about the time and weather in a city."
    ),
    tools=[get_weather, get_current_time],
)

USER_ID = "user_123"
SESSION_ID = "session_abc"
APP_NAME = "weather_agent_team_app"

session_service = InMemorySessionService()
session = await session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID,
)

runner_agent_team = Runner(
    app_name=APP_NAME,
    agent=weather_agent_team,
    session_service=session_service,
)

from google.genai import types

# 假设我们问的问题是 query
content = types.Content(role="user", parts=[types.Part(text=query)])

# 由于方法是 async 的，所以我们得等到所有 events 执行完毕.
# 所以需要在 for 前加上 await 关键词
async for event in runner.run_async(
    user_id=user_id,
    session_id=session_id,
    new_message=content,
):
    # 输出所有中间产物，各种各样的 events
    print(f"  [Event]")
    print(f"\tAuthor: {event.author}")
    print(f"\tType: {type(event).__name__}")
    print(f"\tFinal: {event.is_final_response()}")
    print(f"\tContent: {event.content}")

    # is_final_response() marks the concluding message for the turn.
    if event.is_final_response():
        if event.content and event.content.parts:
            # Assuming text response in the first part
            final_response_text = event.content.parts[0].text
        elif (
            event.actions and event.actions.escalate
        ):  # Handle potential errors/escalations
            final_response_text = (
                f"Agent escalated: {event.error_message or 'No specific message.'}"
            )
        # Add more checks here if needed (e.g., specific error codes)
        break  # Stop processing events once the final response is found

print(f"<<< Agent Response: {final_response_text}")