--- title: Google ADK description: "Track and analyze your Google Agent Development Kit (ADK) AI agents with AgentOps" --- AgentOps provides seamless integration with [Google Agent Development Kit (ADK)](https://google.github.io/adk-docs/), allowing you to track and analyze all your ADK agent interactions automatically. ## Installation ```bash pip pip install agentops google-adk ``` ```bash poetry poetry add agentops google-adk ``` ```bash uv uv pip install agentops google-adk ``` ## Setting Up API Keys Before using Google ADK with AgentOps, you need to set up your API keys. You can obtain: - **GOOGLE_API_KEY**: From the [Google AI Studio](https://aistudio.google.com/app/apikey) - **AGENTOPS_API_KEY**: From your [AgentOps Dashboard](https://app.agentops.ai/) Then to set them up, you can either export them as environment variables or set them in a `.env` file. ```bash Export to CLI export GOOGLE_API_KEY="your_google_api_key_here" export AGENTOPS_API_KEY="your_agentops_api_key_here" ``` ```txt Set in .env file GOOGLE_API_KEY="your_google_api_key_here" AGENTOPS_API_KEY="your_agentops_api_key_here" ``` Then load the environment variables in your Python code: ```python from dotenv import load_dotenv import os # Load environment variables from .env file load_dotenv() # Set up environment variables with fallback values os.environ["GOOGLE_API_KEY"] = os.getenv("GOOGLE_API_KEY") os.environ["AGENTOPS_API_KEY"] = os.getenv("AGENTOPS_API_KEY") ``` ## Usage Initialize AgentOps at the beginning of your application to automatically track all Google ADK agent interactions: ```python import asyncio import json from pydantic import BaseModel, Field import agentops from google.adk.agents import LlmAgent from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.genai import types agentops.init() # --- 1. Define Constants --- APP_NAME = "agent_comparison_app" USER_ID = "test_user_456" SESSION_ID_TOOL_AGENT = "session_tool_agent_xyz" SESSION_ID_SCHEMA_AGENT = "session_schema_agent_xyz" MODEL_NAME = "gemini-2.0-flash" # --- 2. Define Schemas --- # Input schema used by both agents class CountryInput(BaseModel): country: str = Field(description="The country to get information about.") # Output schema ONLY for the second agent class CapitalInfoOutput(BaseModel): capital: str = Field(description="The capital city of the country.") # Note: Population is illustrative; the LLM will infer or estimate this # as it cannot use tools when output_schema is set. population_estimate: str = Field(description="An estimated population of the capital city.") # --- 3. Define the Tool (Only for the first agent) --- def get_capital_city(country: str) -> str: """Retrieves the capital city of a given country.""" print(f"\n-- Tool Call: get_capital_city(country='{country}') --") country_capitals = { "united states": "Washington, D.C.", "canada": "Ottawa", "france": "Paris", "japan": "Tokyo", } result = country_capitals.get(country.lower(), f"Sorry, I couldn't find the capital for {country}.") print(f"-- Tool Result: '{result}' --") return result # --- 4. Configure Agents --- # Agent 1: Uses a tool and output_key capital_agent_with_tool = LlmAgent( model=MODEL_NAME, name="capital_agent_tool", description="Retrieves the capital city using a specific tool.", instruction="""You are a helpful agent that provides the capital city of a country using a tool. The user will provide the country name in a JSON format like {"country": "country_name"}. 1. Extract the country name. 2. Use the `get_capital_city` tool to find the capital. 3. Respond clearly to the user, stating the capital city found by the tool. """, tools=[get_capital_city], input_schema=CountryInput, output_key="capital_tool_result", # Store final text response ) # Agent 2: Uses output_schema (NO tools possible) structured_info_agent_schema = LlmAgent( model=MODEL_NAME, name="structured_info_agent_schema", description="Provides capital and estimated population in a specific JSON format.", instruction=f"""You are an agent that provides country information. The user will provide the country name in a JSON format like {{"country": "country_name"}}. Respond ONLY with a JSON object matching this exact schema: {json.dumps(CapitalInfoOutput.model_json_schema(), indent=2)} Use your knowledge to determine the capital and estimate the population. Do not use any tools. """, # *** NO tools parameter here - using output_schema prevents tool use *** input_schema=CountryInput, output_schema=CapitalInfoOutput, # Enforce JSON output structure output_key="structured_info_result", # Store final JSON response ) # --- 5. Set up Session Management and Runners --- session_service = InMemorySessionService() # Create a runner for EACH agent capital_runner = Runner( agent=capital_agent_with_tool, app_name=APP_NAME, session_service=session_service ) structured_runner = Runner( agent=structured_info_agent_schema, app_name=APP_NAME, session_service=session_service ) # --- 6. Define Agent Interaction Logic --- async def call_agent_and_print( runner_instance: Runner, agent_instance: LlmAgent, session_id: str, query_json: str ): """Sends a query to the specified agent/runner and prints results.""" print(f"\n>>> Calling Agent: '{agent_instance.name}' | Query: {query_json}") user_content = types.Content(role='user', parts=[types.Part(text=query_json)]) final_response_content = "No final response received." async for event in runner_instance.run_async(user_id=USER_ID, session_id=session_id, new_message=user_content): # print(f"Event: {event.type}, Author: {event.author}") # Uncomment for detailed logging if event.is_final_response() and event.content and event.content.parts: # For output_schema, the content is the JSON string itself final_response_content = event.content.parts[0].text print(f"<<< Agent '{agent_instance.name}' Response: {final_response_content}") current_session = await session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=session_id) stored_output = current_session.state.get(agent_instance.output_key) # Pretty print if the stored output looks like JSON (likely from output_schema) print(f"--- Session State ['{agent_instance.output_key}']: ", end="") try: # Attempt to parse and pretty print if it's JSON parsed_output = json.loads(stored_output) print(json.dumps(parsed_output, indent=2)) except (json.JSONDecodeError, TypeError): # Otherwise, print as string print(stored_output) print("-" * 30) # --- 7. Run Interactions --- async def main(): # Create sessions await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID_TOOL_AGENT) await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID_SCHEMA_AGENT) print("--- Testing Agent with Tool ---") await call_agent_and_print(capital_runner, capital_agent_with_tool, SESSION_ID_TOOL_AGENT, '{"country": "France"}') print("\n\n--- Testing Agent with Output Schema (No Tool Use) ---") await call_agent_and_print(structured_runner, structured_info_agent_schema, SESSION_ID_SCHEMA_AGENT, '{"country": "Japan"}') asyncio.run(main()) ``` ## Examples Implement human-in-the-loop approval workflows with Google ADK agents Visit your [AgentOps Dashboard](https://app.agentops.ai) to see detailed traces of your Google ADK agent interactions, tool usage, and session management.