Google ADK is powerful. But there are plenty of situations where you can't or don't want to use it.
Security policies that block third-party frameworks, existing Python codebases that resist new abstractions, or simply the frustration of debugging through multiple layers of framework magic — these are all real reasons to reach for the raw Gemini API.
I've built agents both ways. For one of my personal apps, I went with the raw API specifically because I wanted the startup overhead as low as possible for a serverless deployment. The resulting code was easier to reason about, and when bugs appeared, they were much faster to isolate.
This guide walks through building a custom agent loop from scratch — the kind you'd actually ship to production. Along the way, you'll understand what ADK is doing under the hood, which makes you better at using it when you do choose to.
The Core Agent Loop
An agent is a model in a loop: it decides which tool to call, calls it, observes the result, and continues until it has a final answer. Here's the minimal production-ready implementation:
# agent_core.py
import google.generativeai as genai
from typing import Callable, Any
genai.configure(api_key="YOUR_GEMINI_API_KEY")
class AgentLoop:
"""Minimal production-ready Gemini API agent loop."""
def __init__(
self,
model_name: str = "gemini-2.5-pro",
tools: list[dict] = None,
tool_functions: dict[str, Callable] = None,
max_iterations: int = 10, # Prevents infinite loops
):
self.model = genai.GenerativeModel(
model_name=model_name,
tools=tools or [],
)
self.tool_functions = tool_functions or {}
self.max_iterations = max_iterations
def run(self, user_message: str) -> str:
"""Run the agent loop and return the final answer."""
chat = self.model.start_chat()
response = chat.send_message(user_message)
for iteration in range(self.max_iterations):
# 1. Check if the model is requesting tool calls
parts = response.candidates[0].content.parts if response.candidates else []
tool_calls = [
p for p in parts
if hasattr(p, "function_call") and p.function_call.name
]
# 2. No tool calls means we have a final answer
if not tool_calls:
return response.text
# 3. Execute all requested tools
tool_results = []
for call in tool_calls:
result = self._execute_tool(
call.function_call.name,
dict(call.function_call.args),
)
tool_results.append({
"function_response": {
"name": call.function_call.name,
"response": {"result": str(result)},
}
})
# 4. Send results back and continue the loop
response = chat.send_message(tool_results)
# Graceful exit if max_iterations is reached
return response.text if hasattr(response, "text") else "Agent reached maximum iterations"
def _execute_tool(self, name: str, args: dict) -> Any:
"""Execute a tool function safely."""
if name not in self.tool_functions:
return f"Error: tool '{name}' not found"
try:
return self.tool_functions[name](**args)
except Exception as e:
return f"Tool execution error: {str(e)}"Two decisions here are non-negotiable in production: max_iterations to prevent infinite loops, and wrapping tool execution in try/except so a single broken tool doesn't crash the entire agent session.
For a deep dive into Function Calling foundations, see this guide
Designing Tool Definitions That Actually Work
The quality of your tool definitions directly determines how reliably the model selects and uses them. The description field is not documentation for humans — it's the instruction the model reads when deciding whether to call this tool.
# tools.py
TOOL_DEFINITIONS = [
{
"function_declarations": [
{
"name": "search_documents",
"description": (
"Search internal documents by keyword. "
"Use this when the user is looking for specific information "
"or when a question requires referenced data to answer accurately. "
"Do NOT use for general knowledge questions."
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query. More specific keywords yield better results.",
},
"max_results": {
"type": "integer",
"description": "Maximum results to return. Default 5, maximum 20.",
"default": 5,
},
},
"required": ["query"],
},
},
{
"name": "execute_python_code",
"description": (
"Execute Python code in a sandboxed environment. "
"Use only for calculations, data transformations, or list operations "
"where deterministic execution is required. "
"File system and network access are NOT available."
),
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "Python code to execute. Only standard library modules allowed.",
},
},
"required": ["code"],
},
},
]
}
]
# Tool implementations
import subprocess
import tempfile
import os
def search_documents(query: str, max_results: int = 5) -> list[dict]:
"""Actual search logic — connect to Elasticsearch, Pinecone, etc."""
return [
{"id": "doc-1", "title": "API Guide", "snippet": f"Document related to '{query}'..."}
][:max_results]
def execute_python_code(code: str) -> str:
"""Execute Python in a restricted sandbox."""
blocked = ["os", "subprocess", "socket", "requests", "http"]
for mod in blocked:
if f"import {mod}" in code or f"from {mod}" in code:
return f"Error: '{mod}' module is not permitted"
with tempfile.NamedTemporaryFile(suffix=".py", mode="w", delete=False) as f:
f.write(code)
path = f.name
try:
result = subprocess.run(
["python3", path],
capture_output=True,
text=True,
timeout=10,
)
return result.stdout if result.returncode == 0 else f"Runtime error: {result.stderr}"
except subprocess.TimeoutExpired:
return "Error: execution timed out (10s limit)"
finally:
os.unlink(path)
TOOL_FUNCTIONS = {
"search_documents": search_documents,
"execute_python_code": execute_python_code,
}The key pattern in the execute_python_code description — "File system and network access are NOT available" — tells the model upfront what this tool can't do. This prevents the model from trying to use it for tasks that will fail, which reduces unnecessary tool calls and saves tokens.
Conversation Memory: Sliding Window + Summarization
The default Chat session manages history automatically, but in production you'll hit two problems quickly: context windows fill up during long conversations, and you need memory to persist across sessions.
# memory.py
import google.generativeai as genai
import json
import time
from typing import Optional
from dataclasses import dataclass, field
@dataclass
class ConversationTurn:
role: str # "user" or "model"
content: str
timestamp: float = field(default_factory=time.time)
class SlidingWindowMemory:
"""
Sliding window conversation memory with automatic summarization.
Compresses old turns to stay within token limits while preserving context.
"""
def __init__(
self,
max_turns: int = 20,
summarize_threshold: int = 15,
model_name: str = "gemini-2.5-pro",
):
self.max_turns = max_turns
self.summarize_threshold = summarize_threshold
self.model_name = model_name
self.turns: list[ConversationTurn] = []
self.summary: Optional[str] = None
def add_turn(self, role: str, content: str):
self.turns.append(ConversationTurn(role=role, content=content))
if len(self.turns) >= self.summarize_threshold:
self._compress_old_turns()
def _compress_old_turns(self):
"""Summarize the older half of the conversation to reduce token usage."""
compress_count = len(self.turns) // 2
turns_to_compress = self.turns[:compress_count]
conversation_text = "\n".join([
f"{t.role}: {t.content}" for t in turns_to_compress
])
model = genai.GenerativeModel(self.model_name)
prompt = (
f"Summarize the following conversation in 3-5 sentences. "
f"Preserve all important facts, decisions, and context:\n\n{conversation_text}"
)
try:
response = model.generate_content(prompt)
new_summary = response.text
self.summary = (
f"{self.summary}\n\n[Additional summary] {new_summary}"
if self.summary else new_summary
)
self.turns = self.turns[compress_count:]
except Exception:
# If summarization fails, drop old turns rather than crash
self.turns = self.turns[compress_count:]
def get_history_for_api(self) -> list[dict]:
"""Convert memory to the format expected by Gemini API chat history."""
history = []
if self.summary:
history.extend([
{"role": "user", "parts": [f"[Conversation summary] {self.summary}"]},
{"role": "model", "parts": ["Understood. I'll keep this context in mind."]},
])
for turn in self.turns:
history.append({"role": turn.role, "parts": [turn.content]})
return history
def to_json(self) -> str:
"""Serialize for persistence (Redis, database, etc.)."""
return json.dumps({
"summary": self.summary,
"turns": [
{"role": t.role, "content": t.content, "timestamp": t.timestamp}
for t in self.turns
],
})
@classmethod
def from_json(cls, json_str: str) -> "SlidingWindowMemory":
"""Restore from serialized state."""
data = json.loads(json_str)
memory = cls()
memory.summary = data.get("summary")
memory.turns = [ConversationTurn(**t) for t in data.get("turns", [])]
return memoryNotice the failure mode in _compress_old_turns: if the summarization API call fails, we drop old turns rather than raising an exception. This is an intentional tradeoff — losing some conversation history is acceptable, but crashing the agent session is not.
Parallel Agent Execution
For tasks that can be decomposed — researching multiple topics simultaneously, analyzing several documents in parallel — concurrent execution dramatically reduces wall-clock time.
# parallel_agent.py
import asyncio
import google.generativeai as genai
from typing import Any
import time
class ParallelAgentOrchestrator:
"""
Runs multiple sub-agents concurrently using asyncio.
Use the Flash model for sub-agents to keep parallel execution cost-effective.
"""
def __init__(self, model_name: str = "gemini-2.5-flash"):
self.model_name = model_name
async def run_subagent(
self,
task_name: str,
task_prompt: str,
tools: list[dict],
tool_functions: dict,
) -> dict[str, Any]:
"""Run a single sub-agent asynchronously."""
start = time.time()
try:
model = genai.GenerativeModel(model_name=self.model_name, tools=tools)
chat = model.start_chat()
response = chat.send_message(task_prompt)
for _ in range(5):
parts = response.candidates[0].content.parts if response.candidates else []
calls = [p for p in parts if hasattr(p, "function_call") and p.function_call.name]
if not calls:
break
results = []
for call in calls:
func = tool_functions.get(call.function_call.name)
result = func(**dict(call.function_call.args)) if func else f"Undefined tool: {call.function_call.name}"
results.append({
"function_response": {
"name": call.function_call.name,
"response": {"result": str(result)},
}
})
response = chat.send_message(results)
return {
"task_name": task_name,
"result": response.text,
"elapsed": time.time() - start,
"status": "success",
}
except Exception as e:
return {
"task_name": task_name,
"result": None,
"elapsed": time.time() - start,
"status": "error",
"error": str(e),
}
async def run_parallel(self, subtasks: list[dict]) -> list[dict]:
"""
Execute multiple subtasks concurrently.
Expected subtask format:
{
"name": "task_identifier",
"prompt": "What to accomplish",
"tools": [...],
"tool_functions": {...},
}
"""
coroutines = [
self.run_subagent(
task_name=t["name"],
task_prompt=t["prompt"],
tools=t.get("tools", []),
tool_functions=t.get("tool_functions", {}),
)
for t in subtasks
]
results = await asyncio.gather(*coroutines, return_exceptions=False)
return list(results)
# Practical example: parallel competitive research
async def competitive_analysis():
orchestrator = ParallelAgentOrchestrator()
subtasks = [
{"name": "product_a", "prompt": "Research Product A: pricing, features, and reviews", "tools": TOOL_DEFINITIONS, "tool_functions": TOOL_FUNCTIONS},
{"name": "product_b", "prompt": "Research Product B: pricing, features, and reviews", "tools": TOOL_DEFINITIONS, "tool_functions": TOOL_FUNCTIONS},
{"name": "product_c", "prompt": "Research Product C: pricing, features, and reviews", "tools": TOOL_DEFINITIONS, "tool_functions": TOOL_FUNCTIONS},
]
results = await orchestrator.run_parallel(subtasks)
# Aggregate results for a synthesis pass
combined = "\n\n".join([
f"[{r['task_name']}]\n{r['result']}"
for r in results if r["status"] == "success"
])
return combined
if __name__ == "__main__":
print(asyncio.run(competitive_analysis()))One important caveat: running 3–5 agents in parallel multiplies your API request rate proportionally. On the free tier, this will hit rate limits quickly. Add an asyncio.Semaphore to cap concurrency if you're operating near quota limits.
Parallel Function Calling production patterns in depth
Error Recovery and Retry Design
Production agents fail. 503s, rate limits, network timeouts — they're inevitable. Your retry logic determines whether users see errors or just a slight delay.
# retry.py
import google.generativeai as genai
from google.api_core import exceptions as google_exceptions
import time
import random
from typing import Callable, TypeVar
from functools import wraps
T = TypeVar("T")
class AgentRetryPolicy:
"""
Exponential backoff with jitter for Gemini API calls.
Why jitter matters: without it, all concurrent agents that fail
simultaneously will retry at the same moment, causing another failure wave.
Adding random(0, 1) spreads retries across time.
"""
def __init__(
self,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.retryable = (
google_exceptions.ResourceExhausted, # 429
google_exceptions.ServiceUnavailable, # 503
google_exceptions.DeadlineExceeded, # timeout
ConnectionError,
TimeoutError,
)
def execute(self, func: Callable[..., T], *args, **kwargs) -> T:
"""
Execute func with retry. On success, returns immediately.
On exhaustion, re-raises the final exception.
Example behavior:
- Attempt 1 fails (503) → wait ~1.2s → retry
- Attempt 2 fails (503) → wait ~2.4s → retry
- Attempt 3 succeeds → returns result
"""
last_exc = None
for attempt in range(self.max_retries + 1):
try:
return func(*args, **kwargs)
except self.retryable as e:
last_exc = e
if attempt == self.max_retries:
break
delay = min(
self.base_delay * (2 ** attempt) + random.uniform(0, 1),
self.max_delay,
)
print(f"Attempt {attempt + 1}/{self.max_retries} failed ({type(e).__name__}). Retrying in {delay:.1f}s...")
time.sleep(delay)
except Exception:
raise # Non-retryable: fail immediately
raise last_exc
def with_retry(max_retries: int = 3, base_delay: float = 1.0):
"""Decorator for adding retry behavior to any function."""
def decorator(func: Callable) -> Callable:
policy = AgentRetryPolicy(max_retries=max_retries, base_delay=base_delay)
@wraps(func)
def wrapper(*args, **kwargs):
return policy.execute(func, *args, **kwargs)
return wrapper
return decorator
# Usage example
class ResilientAgent:
def __init__(self):
self.model = genai.GenerativeModel("gemini-2.5-pro")
self.retry = AgentRetryPolicy(max_retries=3)
def run(self, prompt: str) -> str:
chat = self.model.start_chat()
response = self.retry.execute(chat.send_message, prompt)
return response.textThe thundering herd problem is subtle but real in production. I've seen staging environments where a single upstream slowdown caused all agent instances to retry in lockstep, amplifying the load spike. The random.uniform(0, 1) addition is the simplest fix.
Detailed error handling and rate limit patterns for production
Common Pitfalls and How to Avoid Them
Building agents without a framework surfaces failure modes that ADK handles silently. Here are the ones that will definitely catch you.
Pitfall 1: Wrong function_response format
This is the most common issue. The API silently ignores malformed tool results, causing the model to hallucinate or stall.
# ❌ Wrong: string directly as result
result = "Search found 3 documents"
# ❌ Wrong: missing "response" key
result = {
"function_response": {
"name": "search_documents",
"content": "Search found 3 documents", # Wrong key
}
}
# ✅ Correct format
result = {
"function_response": {
"name": "search_documents", # Must match definition exactly
"response": { # Required key
"result": "Search found 3 documents",
},
}
}Pitfall 2: Tool name case sensitivity
The name in your tool definition and the name in function_call.name must match character-for-character. Always validate before dispatching:
tool_name = part.function_call.name
if tool_name in self.tool_functions:
result = self.tool_functions[tool_name](**args)
else:
result = f"Unknown tool requested: {tool_name}" # Never silently ignore thisPitfall 3: Not checking for empty parts
# ❌ Crashes when parts is empty
first_part = response.candidates[0].content.parts[0]
# ✅ Safe
parts = response.candidates[0].content.parts if response.candidates else []
tool_calls = [p for p in parts if hasattr(p, "function_call") and p.function_call.name]
if not tool_calls:
return response.text or "No response generated"Pitfall 4: Using Pro for everything
Running every agent step through Gemini 2.5 Pro will blow your budget faster than you expect. A tiered model strategy keeps costs manageable:
ORCHESTRATOR_MODEL = "gemini-2.5-pro" # Final synthesis, complex reasoning
SUBAGENT_MODEL = "gemini-2.5-flash" # Tool execution, intermediate steps
SUMMARIZATION_MODEL = "gemini-2.5-flash" # Memory compression, classificationWhen to Use ADK vs. Custom Implementation
Rather than one being "better," they optimize for different things.
ADK makes sense when:
- You're working on a team and want standardized patterns
- Deep Google Cloud integration is important
- You want to move quickly without reinventing patterns
- The agent design space is still being explored
Custom implementation makes sense when:
- You're integrating into an existing Python codebase
- You need predictable behavior across framework version updates
- Debuggability is a priority (stack traces go straight to your code)
- You're deploying to lightweight environments (Cloudflare Workers, Lambda)
- You need optimization strategies ADK doesn't expose (custom caching, specialized routing)
The best engineers I've worked with treat ADK and raw API usage as tools with different tradeoffs, not a religious debate.
Monitor your production agents with structured observability ADK-based multi-agent architecture for team environments
Where to Start
Start with the minimal AgentLoop class, connect one real tool from your project, and run it. The first time the model calls your tool and uses the result to generate a response, it clicks.
From there, add error handling, then memory, then parallelism — in that order. Each layer solves a specific production problem, and adding them incrementally keeps the system understandable.
Building a Complete Working Agent: End-to-End Example
Theory is useful, but seeing all the pieces connected is more valuable. Here's a complete "research assistant" agent that combines everything covered so far — tool calling, memory, retry logic, and cost-aware model selection.
# research_agent.py
# Complete implementation: research assistant with memory and retry
import google.generativeai as genai
import asyncio
import json
import time
import random
from dataclasses import dataclass, field
from typing import Any, Optional
from google.api_core import exceptions as google_exceptions
genai.configure(api_key="YOUR_GEMINI_API_KEY")
# ---- Tool Definitions ----
RESEARCH_TOOLS = [
{
"function_declarations": [
{
"name": "fetch_webpage_content",
"description": (
"Retrieve the text content of a webpage by URL. "
"Use when you need current information from a specific source. "
"Returns the page's main text content only."
),
"parameters": {
"type": "object",
"properties": {
"url": {"type": "string", "description": "The URL to fetch."},
},
"required": ["url"],
},
},
{
"name": "calculate",
"description": (
"Evaluate a mathematical expression. "
"Use for any arithmetic, percentages, or unit conversions. "
"Input must be a valid Python expression."
),
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Python math expression, e.g., '(42 * 1.08) / 12'",
},
},
"required": ["expression"],
},
},
]
}
]
def fetch_webpage_content(url: str) -> str:
"""Fetch webpage content. Replace with your actual fetching logic."""
# In production: use requests + BeautifulSoup or a search API
return f"[Simulated content from {url}]"
def calculate(expression: str) -> str:
"""Safely evaluate a math expression."""
# Block anything that isn't arithmetic
allowed_chars = set("0123456789+-*/()., ")
if not all(c in allowed_chars for c in expression):
return "Error: only basic arithmetic is permitted"
try:
result = eval(expression, {"__builtins__": {}}, {})
return str(result)
except Exception as e:
return f"Calculation error: {str(e)}"
RESEARCH_TOOL_FUNCTIONS = {
"fetch_webpage_content": fetch_webpage_content,
"calculate": calculate,
}
# ---- Memory ----
@dataclass
class Turn:
role: str
content: str
timestamp: float = field(default_factory=time.time)
class ConversationMemory:
def __init__(self, max_turns: int = 16):
self.max_turns = max_turns
self.turns: list[Turn] = []
self.session_summary: Optional[str] = None
def add(self, role: str, content: str):
self.turns.append(Turn(role=role, content=content))
if len(self.turns) > self.max_turns:
self._summarize_and_trim()
def _summarize_and_trim(self):
half = len(self.turns) // 2
to_compress = self.turns[:half]
text = "\n".join(f"{t.role}: {t.content[:200]}" for t in to_compress)
model = genai.GenerativeModel("gemini-2.5-flash")
try:
resp = model.generate_content(f"Summarize this conversation in 3 sentences:\n{text}")
self.session_summary = (
f"{self.session_summary} | {resp.text}" if self.session_summary else resp.text
)
except Exception:
pass
self.turns = self.turns[half:]
def to_api_history(self) -> list[dict]:
history = []
if self.session_summary:
history += [
{"role": "user", "parts": [f"[Prior context] {self.session_summary}"]},
{"role": "model", "parts": ["Context noted."]},
]
for t in self.turns:
history.append({"role": t.role, "parts": [t.content]})
return history
def save(self, path: str):
with open(path, "w") as f:
json.dump({
"summary": self.session_summary,
"turns": [{"role": t.role, "content": t.content} for t in self.turns],
}, f, ensure_ascii=False, indent=2)
@classmethod
def load(cls, path: str) -> "ConversationMemory":
with open(path) as f:
data = json.load(f)
mem = cls()
mem.session_summary = data.get("summary")
mem.turns = [Turn(**t) for t in data.get("turns", [])]
return mem
# ---- Retry ----
def call_with_retry(func, *args, max_retries=3, **kwargs):
retryable = (
google_exceptions.ResourceExhausted,
google_exceptions.ServiceUnavailable,
google_exceptions.DeadlineExceeded,
)
last = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except retryable as e:
last = e
if attempt == max_retries:
break
delay = (2 ** attempt) + random.uniform(0, 0.5)
time.sleep(delay)
except Exception:
raise
raise last
# ---- Research Agent ----
class ResearchAgent:
"""
A complete research assistant agent with memory, retry, and cost controls.
Usage:
agent = ResearchAgent()
answer = agent.ask("Compare the pricing of Gemini 2.5 Pro and GPT-4o")
print(answer)
"""
def __init__(self, memory_path: Optional[str] = None):
self.model = genai.GenerativeModel(
model_name="gemini-2.5-pro",
tools=RESEARCH_TOOLS,
system_instruction=(
"You are a precise research assistant. "
"Always cite which tool results you used to form your answer. "
"If you are uncertain, say so explicitly rather than guessing."
),
)
self.memory = ConversationMemory.load(memory_path) if memory_path else ConversationMemory()
self.memory_path = memory_path
self.total_tool_calls = 0
self.max_tool_calls_per_turn = 15 # Hard limit per question
def ask(self, question: str) -> str:
"""Ask the agent a question and return its answer."""
# Restore conversation history
history = self.memory.to_api_history()
chat = self.model.start_chat(history=history)
response = call_with_retry(chat.send_message, question)
self.memory.add("user", question)
tool_calls_this_turn = 0
for _ in range(10):
parts = response.candidates[0].content.parts if response.candidates else []
calls = [p for p in parts if hasattr(p, "function_call") and p.function_call.name]
if not calls:
break
# Safety: prevent runaway tool loops
tool_calls_this_turn += len(calls)
if tool_calls_this_turn > self.max_tool_calls_per_turn:
break
results = []
for call in calls:
func = RESEARCH_TOOL_FUNCTIONS.get(call.function_call.name)
if func:
try:
result = func(**dict(call.function_call.args))
except Exception as e:
result = f"Tool error: {e}"
else:
result = f"Unknown tool: {call.function_call.name}"
results.append({
"function_response": {
"name": call.function_call.name,
"response": {"result": str(result)},
}
})
self.total_tool_calls += 1
response = call_with_retry(chat.send_message, results)
answer = response.text or "Unable to generate a response."
self.memory.add("model", answer)
# Persist memory if path is set
if self.memory_path:
self.memory.save(self.memory_path)
return answer
# Example usage
if __name__ == "__main__":
agent = ResearchAgent(memory_path="./session.json")
# First question
print(agent.ask("What are the key differences between Gemini API and Vertex AI Gemini?"))
# Follow-up uses conversation memory
print(agent.ask("Which one should I use for a startup with limited budget?"))
print(f"\nTotal tool calls this session: {agent.total_tool_calls}")The max_tool_calls_per_turn limit deserves explanation. Occasionally, a model will enter a pattern where it keeps calling tools with slightly different arguments, never quite getting to a final answer. Setting a hard cap per question (not just per loop iteration) ensures this edge case terminates gracefully.
Testing Your Agent
Testing agent behavior is fundamentally different from testing deterministic code. The model's responses vary, so you test for properties rather than exact outputs.
# test_agent.py
import pytest
from unittest.mock import patch, MagicMock
class TestAgentToolRouting:
"""Test that the agent routes to the correct tools."""
def test_math_question_uses_calculate_tool(self):
"""Agent should use calculate tool for arithmetic questions."""
call_log = []
original_calculate = calculate
def tracked_calculate(expression: str) -> str:
call_log.append(("calculate", expression))
return original_calculate(expression)
# Temporarily replace the tool function
RESEARCH_TOOL_FUNCTIONS["calculate"] = tracked_calculate
agent = ResearchAgent()
agent.ask("What is 15% of 2,500?")
# Verify the calculate tool was called
assert any(name == "calculate" for name, _ in call_log), \
"Expected 'calculate' tool to be called for arithmetic question"
# Restore original
RESEARCH_TOOL_FUNCTIONS["calculate"] = original_calculate
def test_tool_error_does_not_crash_agent(self):
"""Agent should recover gracefully when a tool raises an exception."""
def broken_tool(**kwargs):
raise RuntimeError("Simulated tool failure")
RESEARCH_TOOL_FUNCTIONS["calculate"] = broken_tool
agent = ResearchAgent()
try:
result = agent.ask("Calculate 100 + 200")
# Should return some response, not raise
assert result is not None
assert len(result) > 0
finally:
RESEARCH_TOOL_FUNCTIONS["calculate"] = calculate
def test_max_tool_calls_limit_respected(self):
"""Agent should stop calling tools after hitting the per-turn limit."""
call_count = [0]
def counting_tool(**kwargs):
call_count[0] += 1
return "partial result"
RESEARCH_TOOL_FUNCTIONS["fetch_webpage_content"] = counting_tool
agent = ResearchAgent()
agent.max_tool_calls_per_turn = 3 # Set low for testing
agent.ask("Research these 10 URLs and compare them: ...")
assert call_count[0] <= 3, \
f"Expected max 3 tool calls, got {call_count[0]}"
class TestConversationMemory:
"""Test memory serialization and compression."""
def test_memory_round_trip(self, tmp_path):
"""Memory should survive save/load cycle."""
mem = ConversationMemory()
mem.add("user", "Hello, how are you?")
mem.add("model", "I'm doing well, thanks for asking!")
save_path = str(tmp_path / "memory.json")
mem.save(save_path)
restored = ConversationMemory.load(save_path)
assert len(restored.turns) == 2
assert restored.turns[0].role == "user"
assert "Hello" in restored.turns[0].content
def test_compression_triggers_at_threshold(self):
"""Memory should compress when turn count exceeds max_turns."""
mem = ConversationMemory(max_turns=4)
for i in range(6):
mem.add("user", f"Question {i}")
mem.add("model", f"Answer {i}")
# After compression, turn count should be at or below max
assert len(mem.turns) <= 4Writing tests like these catches behavioral regressions when you update tools or switch models. The property-based assertion style — "calculate tool was called at least once" rather than "calculate tool was called with exactly '15/100 * 2500'" — keeps tests robust against normal model output variation.
Monitoring and Cost Tracking in Production
Without visibility into how your agent is behaving, debugging production issues is guesswork.
# monitoring.py
import time
import json
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime, timezone
@dataclass
class AgentRun:
session_id: str
question: str
answer: str
tool_calls: list[dict]
model_name: str
elapsed_seconds: float
timestamp: str = field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
error: Optional[str] = None
def estimated_cost_usd(self) -> float:
"""
Rough cost estimate based on approximate token counts.
Update pricing constants as Gemini pricing changes.
Gemini 2.5 Pro (approximate as of 2026-04):
- Input: $1.25 / 1M tokens (≤200K context)
- Output: $10.00 / 1M tokens
"""
# Rough estimates: 4 chars per token
input_tokens = len(self.question) / 4
output_tokens = len(self.answer) / 4
if "flash" in self.model_name:
# Gemini 2.5 Flash approximate pricing
return (input_tokens * 0.0000001875) + (output_tokens * 0.0000007500)
else:
# Gemini 2.5 Pro
return (input_tokens * 0.00000125) + (output_tokens * 0.0000100)
def to_log_entry(self) -> str:
return json.dumps({
"session_id": self.session_id,
"timestamp": self.timestamp,
"question_preview": self.question[:100],
"tool_calls": len(self.tool_calls),
"tools_used": list({c.get("name") for c in self.tool_calls}),
"elapsed_s": round(self.elapsed_seconds, 2),
"estimated_cost_usd": round(self.estimated_cost_usd(), 6),
"model": self.model_name,
"error": self.error,
}, ensure_ascii=False)
class InstrumentedAgent(ResearchAgent):
"""ResearchAgent with built-in cost and performance tracking."""
def __init__(self, log_path: str = "./agent_runs.jsonl", **kwargs):
super().__init__(**kwargs)
self.log_path = log_path
self.run_history: list[AgentRun] = []
def ask(self, question: str) -> str:
start = time.time()
tool_calls_record = []
# Temporarily wrap tool functions to record calls
original_functions = dict(RESEARCH_TOOL_FUNCTIONS)
for name, func in original_functions.items():
def make_tracked(n, f):
def tracked(**kwargs):
result = f(**kwargs)
tool_calls_record.append({"name": n, "args": kwargs})
return result
return tracked
RESEARCH_TOOL_FUNCTIONS[name] = make_tracked(name, func)
error = None
try:
answer = super().ask(question)
except Exception as e:
answer = f"Error: {e}"
error = str(e)
finally:
RESEARCH_TOOL_FUNCTIONS.update(original_functions)
run = AgentRun(
session_id=id(self.memory).__str__(),
question=question,
answer=answer,
tool_calls=tool_calls_record,
model_name="gemini-2.5-pro",
elapsed_seconds=time.time() - start,
error=error,
)
self.run_history.append(run)
with open(self.log_path, "a", encoding="utf-8") as f:
f.write(run.to_log_entry() + "\n")
return answer
def cost_summary(self) -> dict:
"""Return aggregate cost and performance stats for this session."""
if not self.run_history:
return {}
total_cost = sum(r.estimated_cost_usd() for r in self.run_history)
avg_elapsed = sum(r.elapsed_seconds for r in self.run_history) / len(self.run_history)
total_tools = sum(len(r.tool_calls) for r in self.run_history)
return {
"total_runs": len(self.run_history),
"total_estimated_cost_usd": round(total_cost, 4),
"avg_elapsed_seconds": round(avg_elapsed, 2),
"total_tool_calls": total_tools,
"avg_tools_per_run": round(total_tools / len(self.run_history), 1),
}Logging to a .jsonl file (one JSON object per line) makes it easy to pipe into analytics tools, grep for specific sessions, or ingest into a dashboard later. The estimated cost tracking gives you an early warning before a billing surprise at end of month.
Production observability setup for Gemini API agents
ADK vs. Custom: An Honest Assessment
After building agents both ways across several projects, here's what I've actually observed:
ADK advantages that matter:
- Automatic agent state management (you don't think about chat history at all)
- Built-in support for A2A (Agent-to-Agent) protocol
- Google Cloud tracing integration out of the box
- Faster to prototype when you just want to see if an idea works
Custom implementation advantages that matter:
- Predictable breakpoints when debugging (the call stack is your own code)
- No surprise behavior changes when ADK releases a new version
- You control exactly what gets logged (important for compliance)
- Works in environments where Python package size matters (Lambda layers, edge runtimes)
- You can add caching, routing, and fallback logic at any layer without fighting the framework
For a new project where I'm still figuring out what the agent needs to do, I'd start with ADK. For a production system where I know the requirements and need to own every inch of the behavior, I'd go custom. Both approaches produce working agents — the choice is really about operational tradeoffs over the life of the project.
The implementation patterns in this guide — retry with jitter, sliding window memory, per-turn tool call limits — translate directly into ADK-based systems too. Understanding them at the raw API level makes you more effective regardless of which abstraction you choose to build on top.