Created
November 7, 2025 05:44
-
-
Save janakiramm/ee328201c37d966d2d4660faeab26711 to your computer and use it in GitHub Desktop.
Building an agent from scratch with Google ADK
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| ## Part 1 - Import modules | |
| import asyncio | |
| import logging | |
| from typing import Dict, TypedDict | |
| from google.adk.agents import LlmAgent | |
| from google.adk.sessions import InMemorySessionService | |
| from google.adk.runners import Runner | |
| from google.genai import types | |
| LOG_LEVEL = logging.getLevelName("ERROR") | |
| logging.basicConfig(level=LOG_LEVEL) | |
| log = logging.getLogger("weather_demo") | |
| ## Part 2 - Define mock API | |
| class WeatherResult(TypedDict, total=False): | |
| status: str | |
| report: str | |
| error_message: str | |
| MOCK_WEATHER_DB: Dict[str, str] = { | |
| "newyork": "The weather in New York is sunny with a temperature of 25°C.", | |
| "london": "It's cloudy in London with a temperature of 15°C.", | |
| "tokyo": "Tokyo is experiencing light rain and a temperature of 18°C.", | |
| } | |
| def get_weather(city: str) -> WeatherResult: | |
| """Look up a short weather report for a supported city (London, New York, Tokyo).""" | |
| log.info("Tool:get_weather called for city=%s", city) | |
| key = "".join(city.split()).casefold() | |
| if key in MOCK_WEATHER_DB: | |
| return {"status": "success", "report": MOCK_WEATHER_DB[key]} | |
| return { | |
| "status": "error", | |
| "error_message": ( | |
| f"Sorry, I don't have weather information for '{city}'. " | |
| f"Try one of: {', '.join(sorted(MOCK_WEATHER_DB))}." | |
| ), | |
| } | |
| ## Part 3 - Define the agent | |
| AGENT_MODEL = "gemini-2.5-flash" | |
| weather_agent = LlmAgent( | |
| name="weather_agent_v1", | |
| model=AGENT_MODEL, | |
| description="Provides weather information for specific cities.", | |
| instruction=( | |
| "You are a helpful weather assistant. " | |
| "Use the 'get_weather' tool to answer city weather questions. " | |
| "If the tool returns an error, explain it politely; otherwise present the report." | |
| ), | |
| tools=[get_weather], | |
| ) | |
| ## Part 4 - Define session and runner | |
| APP_NAME = "weather_tutorial_app" | |
| USER_ID = "user_1" | |
| SESSION_ID = "session_001" | |
| session_service = InMemorySessionService() | |
| runner = Runner(agent=weather_agent, app_name=APP_NAME, session_service=session_service) | |
| ## Part 5 - Helper Function | |
| async def call_agent(query: str) -> None: | |
| content = types.Content(role="user", parts=[types.Part.from_text(text=query)]) | |
| final_text = "(no final response)" | |
| async for event in runner.run_async( | |
| user_id=USER_ID, session_id=SESSION_ID, new_message=content | |
| ): | |
| if event.is_final_response() and event.content and event.content.parts: | |
| final_text = "".join(getattr(p, "text", "") or "" for p in event.content.parts) | |
| print(f">>> {query}\n<<< {final_text}\n") | |
| ## Part 6 - Define and invoke main() | |
| async def main(): | |
| await session_service.create_session( | |
| app_name=APP_NAME, | |
| user_id=USER_ID, | |
| session_id=SESSION_ID, | |
| ) | |
| print(f"Session created: App='{APP_NAME}', User='{USER_ID}', Session='{SESSION_ID}'") | |
| print(f"Runner created for agent '{runner.agent.name}'.\n") | |
| await call_agent("What is the weather like in London?") | |
| await call_agent("How about Paris?") | |
| await call_agent("Tell me the weather in New York") | |
| if __name__ == "__main__": | |
| try: | |
| asyncio.run(main()) | |
| except Exception as e: | |
| log.exception("Unhandled error: %s", e) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment