ADK で毎朝お天気を通知するエージェントを作る

import requests
from datetime import datetime
from typing import Dict, Any

def get_weather_forecast(lat: float = 35.6895, lon: float = 139.6917) -> Dict[str, Any]:
    """
    Open-Meteo APIから天気予報と過去の天気を取得する関数
    デフォルトは東京 (lat=35.6895, lon=139.6917)

    Args:
        lat (float): 緯度
        lon (float): 経度
        
    Returns:
        Dict[str, Any]: 天気情報を含む辞書
    """
    
    # 今日の日付を取得
    today = datetime.now()
    today_str = today.strftime('%Y-%m-%d')
    
    # 天気予報APIエンドポイント
    forecast_url = "https://api.open-meteo.com/v1/forecast"
    
    # パラメータ設定: 最高気温、最低気温、降水確率
    params = {
        "latitude": lat,
        "longitude": lon,
        "daily": ["weathercode", "temperature_2m_max", "temperature_2m_min", "precipitation_probability_max"],
        "timezone": "Asia/Tokyo",
        "past_days": 1 # 過去1日分（昨日）のデータも含める
    }
    
    try:
        response = requests.get(forecast_url, params=params)
        response.raise_for_status()
        data = response.json()
        
        daily = data.get('daily', {})
        time_list = daily.get('time', [])
        
        # 今日のインデックスを探す
        try:
            today_idx = time_list.index(today_str)
        except ValueError:
            # もし日付が見つからない場合は先頭と仮定
            today_idx = 0

        # 昨日のインデックス（なければ None）
        yesterday_idx = today_idx - 1 if today_idx > 0 else None
        
        weather_info = {
            "location": f"Lat: {lat}, Lon: {lon}",
            "today": {
                "date": today_str,
                "max_temp": daily['temperature_2m_max'][today_idx],
                "min_temp": daily['temperature_2m_min'][today_idx],
                "precip_prob": daily['precipitation_probability_max'][today_idx],
                "weather_code": daily['weathercode'][today_idx]
            }
        }
        
        if yesterday_idx is not None and yesterday_idx >= 0:
            weather_info["yesterday"] = {
                "date": time_list[yesterday_idx],
                "max_temp": daily['temperature_2m_max'][yesterday_idx],
                "min_temp": daily['temperature_2m_min'][yesterday_idx],
                "precip_prob": daily['precipitation_probability_max'][yesterday_idx],
                "weather_code": daily['weathercode'][yesterday_idx]
            }
            
            # 変化の計算
            temp_diff_max = weather_info["today"]["max_temp"] - weather_info["yesterday"]["max_temp"]
            temp_diff_min = weather_info["today"]["min_temp"] - weather_info["yesterday"]["min_temp"]
            
            weather_info["comparison"] = {
                "temp_diff_max": round(temp_diff_max, 1),
                "temp_diff_min": round(temp_diff_min, 1)
            }
        else:
            weather_info["comparison"] = "昨日との比較データなし"

        return weather_info

    except Exception as e:
        return {"error": str(e)}

from google.adk import Agent
from .tools.weather_tool import get_weather_forecast

root_agent = Agent(
    model='gemini-2.5-flash',
    name='weather_agent',
    description='A weather assistant that reports daily forecast to Slack.',
    instruction='''
    あなたは天気予報を通知する親切なアシスタントです。ユーザーの要望に対して、必要なツールを使用して回答を作成してください。
    
    また、気象情報については以下の項目を含めるようにしてください。
       - 今日の天気概況（天気、最高気温、最低気温、降水確率）
       - 昨日との気温差（寒くなったか、暑くなったか）
       - 気温に応じた服装へのアドバイス（例：「昨日より寒いのでコートが必要です」など）
       - 傘が必要かどうか
    ''',
    tools=[get_weather_forecast]
)

# プロジェクトルートで実行
uv run adk web .

GOOGLE_GENAI_USE_VERTEXAI=True
GOOGLE_CLOUD_PROJECT={{ プロジェクトID }}
GOOGLE_CLOUD_LOCATION=us-central1
LOCATION=us-central1

+ # Slack Configuration
+ SLACK_BOT_TOKEN={{ 取得したトークン }}
+ SLACK_CHANNEL_ID={{ 通知先チャンネルID }}

import uuid
from fastapi import Request
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.genai.types import Content, Part
from google.adk.cli.fast_api import get_fast_api_app
from weather_agent.agent import root_agent

app = get_fast_api_app(agents_dir=".", web=False)

@app.post("/daily_run")
async def daily_run(request: Request):
    # セッション作成
    session_service = InMemorySessionService()
    runner = Runner(agent=root_agent, app_name="weather_agent", session_service=session_service)
    session_id = str(uuid.uuid4())
    user_id = "scheduler"
    app_name = "weather_agent"
    await session_service.create_session(app_name=app_name, user_id=user_id, session_id=session_id)
    
    # プロンプト設定
    query_text = "東京の今日の天気を調べて、Slackに通知して"
    content = Content(role="user", parts=[Part(text=query_text)])
    
    # Execute agent
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=content
    ):
        pass # Simply consume events to trigger execution
    
    await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session_id)

    return {"status": "ok", "session_id": session_id}