+ "content": "\"\"\"Basic Chat feature.\"\"\"\n\nfrom __future__ import annotations\n\nfrom fastapi import FastAPI\nfrom ag_ui_adk import ADKAgent, add_adk_fastapi_endpoint\nfrom google.adk.agents import LlmAgent\nfrom google.adk import tools as adk_tools\n\ndef get_weather_condition(code: int) -> str:\n \"\"\"Map weather code to human-readable condition.\n\n Args:\n code: WMO weather code.\n\n Returns:\n Human-readable weather condition string.\n \"\"\"\n conditions = {\n 0: \"Clear sky\",\n 1: \"Mainly clear\",\n 2: \"Partly cloudy\",\n 3: \"Overcast\",\n 45: \"Foggy\",\n 48: \"Depositing rime fog\",\n 51: \"Light drizzle\",\n 53: \"Moderate drizzle\",\n 55: \"Dense drizzle\",\n 56: \"Light freezing drizzle\",\n 57: \"Dense freezing drizzle\",\n 61: \"Slight rain\",\n 63: \"Moderate rain\",\n 65: \"Heavy rain\",\n 66: \"Light freezing rain\",\n 67: \"Heavy freezing rain\",\n 71: \"Slight snow fall\",\n 73: \"Moderate snow fall\",\n 75: \"Heavy snow fall\",\n 77: \"Snow grains\",\n 80: \"Slight rain showers\",\n 81: \"Moderate rain showers\",\n 82: \"Violent rain showers\",\n 85: \"Slight snow showers\",\n 86: \"Heavy snow showers\",\n 95: \"Thunderstorm\",\n 96: \"Thunderstorm with slight hail\",\n 99: \"Thunderstorm with heavy hail\",\n }\n return conditions.get(code, \"Unknown\")\n\n\nasync def get_weather(location: str) -> str:\n \"\"\"Get current weather for a location.\n\n Args:\n location: City name.\n\n Returns:\n Dictionary with weather information including temperature, feels like,\n humidity, wind speed, wind gust, conditions, and location name.\n \"\"\"\n async with httpx.AsyncClient() as client:\n # Geocode the location\n geocoding_url = (\n f\"https://geocoding-api.open-meteo.com/v1/search?name={location}&count=1\"\n )\n geocoding_response = await client.get(geocoding_url)\n geocoding_data = geocoding_response.json()\n\n if not geocoding_data.get(\"results\"):\n raise ValueError(f\"Location '{location}' not found\")\n\n result = geocoding_data[\"results\"][0]\n latitude = result[\"latitude\"]\n longitude = result[\"longitude\"]\n name = result[\"name\"]\n\n # Get weather data\n weather_url = (\n f\"https://api.open-meteo.com/v1/forecast?\"\n f\"latitude={latitude}&longitude={longitude}\"\n f\"¤t=temperature_2m,apparent_temperature,relative_humidity_2m,\"\n f\"wind_speed_10m,wind_gusts_10m,weather_code\"\n )\n weather_response = await client.get(weather_url)\n weather_data = weather_response.json()\n\n current = weather_data[\"current\"]\n\n return json.dumps({\n \"temperature\": current[\"temperature_2m\"],\n \"feelsLike\": current[\"apparent_temperature\"],\n \"humidity\": current[\"relative_humidity_2m\"],\n \"windSpeed\": current[\"wind_speed_10m\"],\n \"windGust\": current[\"wind_gusts_10m\"],\n \"conditions\": get_weather_condition(current[\"weather_code\"]),\n \"location\": name,\n })\n\n# Create a sample ADK agent (this would be your actual agent)\nsample_agent = LlmAgent(\n name=\"assistant\",\n model=\"gemini-2.0-flash\",\n instruction=\"\"\"\n You are a helpful weather assistant that provides accurate weather information.\n\n Your primary function is to help users get weather details for specific locations. When responding:\n - Always ask for a location if none is provided\n - If the location name isn’t in English, please translate it\n - If giving a location with multiple parts (e.g. \"New York, NY\"), use the most relevant part (e.g. \"New York\")\n - Include relevant details like humidity, wind conditions, and precipitation\n - Keep responses concise but informative\n\n Use the get_weather tool to fetch current weather data.\n \"\"\",\n tools=[adk_tools.preload_memory_tool.PreloadMemoryTool(), get_weather]\n)\n\n# Create ADK middleware agent instance\nchat_agent = ADKAgent(\n adk_agent=sample_agent,\n app_name=\"demo_app\",\n user_id=\"demo_user\",\n session_timeout_seconds=3600,\n use_in_memory_services=True\n)\n\n# Create FastAPI app\napp = FastAPI(title=\"ADK Middleware Weather Agent\")\n\n# Add the ADK endpoint\nadd_adk_fastapi_endpoint(app, chat_agent, path=\"/\")\n",
0 commit comments