Google ADK Callbacks & Guardrails: A Complete Production Guide to Agent Monitoring and Safety Control

Why Uncontrollable AI Agents Are the Biggest Production Risk

You've built a Google ADK agent, confirmed it works locally, and passed your tests. Then you deploy to production — and things go sideways.

Here are scenarios that actually happen. A user asks the agent to "research competitor pricing." The agent generates tool calls attempting to access internal competitor systems. A cost-estimation agent enters an infinite loop, consuming thousands of dollars worth of API tokens in a single request. A customer support agent spontaneously promises a refund it has no authority to make.

These are documented production failure patterns. Gemini API quality improves every year, but the probabilistic nature of LLMs means edge-case misbehavior can never be fully eliminated through prompting alone.

Google ADK addresses this with two mechanisms: Callbacks and Guardrails. Callbacks are hooks that let you inject custom functions at specific stages of agent execution. Guardrails automatically inspect and filter inputs and outputs. Together, they let you see what the agent is trying to do at every step — and physically prevent it from doing things it shouldn't.

This guide walks through production implementation patterns for both, using real failure scenarios as context, with verified working code throughout.

Understanding the ADK Callback Structure

When Callbacks Fire

ADK Callbacks attach to specific points in the agent execution lifecycle. As of ADK v1.x (April 2026):

Agent-level

before_agent_callback: Fires before the agent begins processing
after_agent_callback: Fires after the agent finishes

Model call level

before_model_callback: Fires before the request is sent to Gemini
after_model_callback: Fires after the Gemini response is received

Tool call level

before_tool_callback: Fires before a tool executes
after_tool_callback: Fires after a tool returns its result

These give you full visibility into the agent's execution flow.

Basic Callback Implementation

Start with a simple logging callback before adding any blocking logic:

import time
import logging
from typing import Any, Optional
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import FunctionTool
from google.genai import types
 
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("adk.monitoring")
 
def log_before_model(
    callback_context: Any,
    llm_request: types.GenerateContentRequest
) -> Optional[types.GenerateContentResponse]:
    """
    Log request details before the model is called.
    Return None to continue normally.
    Return a GenerateContentResponse to skip the model call entirely.
    """
    contents_preview = str(llm_request.contents)[:200]
    logger.info(
        f"[BEFORE_MODEL] agent={callback_context.agent_name} "
        f"contents_preview={contents_preview}"
    )
    # Record start time for latency tracking
    callback_context.state["_model_call_start"] = time.time()
    return None  # Continue normally
 
def log_after_model(
    callback_context: Any,
    llm_response: types.GenerateContentResponse
) -> Optional[types.GenerateContentResponse]:
    """Log latency and token usage after the model responds."""
    start_time = callback_context.state.get("_model_call_start", time.time())
    elapsed_ms = int((time.time() - start_time) * 1000)
 
    usage = llm_response.usage_metadata
    prompt_tokens = usage.prompt_token_count if usage else 0
    output_tokens = usage.candidates_token_count if usage else 0
 
    logger.info(
        f"[AFTER_MODEL] agent={callback_context.agent_name} "
        f"latency_ms={elapsed_ms} "
        f"prompt_tokens={prompt_tokens} "
        f"output_tokens={output_tokens}"
    )
    return None  # Use the response as-is
 
agent = LlmAgent(
    name="monitored_agent",
    model="gemini-2.5-pro",
    instruction="You are a helpful assistant.",
    before_model_callback=log_before_model,
    after_model_callback=log_after_model,
)
 
# Expected log output:
# 2026-04-14 10:45:01 [INFO] adk.monitoring: [BEFORE_MODEL] agent=monitored_agent ...
# 2026-04-14 10:45:03 [INFO] adk.monitoring: [AFTER_MODEL] agent=monitored_agent latency_ms=1842 ...

The return value semantics are critical. In before_model_callback, returning None lets the model call proceed. Returning a GenerateContentResponse skips the actual model call and uses that response instead — this is the foundation of Guardrails.

Implementing Safety Guardrails

Input Guardrail: Block Harmful Requests Before They Reach the Model

import re
from google.genai import types
 
BLOCKED_PATTERNS = [
    r"password|credential|クレデンシャル",
    r"unauthorized.*access|不正アクセス|hack",
    r"delete.*personal.*data|個人情報.*削除",
    r"refund|compensation|返金|補償",  # Promises the agent can't make
]
 
def input_guardrail(
    callback_context: Any,
    llm_request: types.GenerateContentRequest
) -> Optional[types.GenerateContentResponse]:
    """
    Analyze the request for dangerous patterns.
    Block the model call and return a safe response if found.
    """
    user_messages = []
    for content in llm_request.contents:
        if content.role == "user":
            for part in content.parts:
                if hasattr(part, "text") and part.text:
                    user_messages.append(part.text)
 
    full_input = " ".join(user_messages).lower()
 
    for pattern in BLOCKED_PATTERNS:
        if re.search(pattern, full_input, re.IGNORECASE):
            logger.warning(
                f"[GUARDRAIL_BLOCKED] agent={callback_context.agent_name} "
                f"pattern={pattern} input_preview={full_input[:100]}"
            )
            return types.GenerateContentResponse(
                candidates=[
                    types.Candidate(
                        content=types.Content(
                            role="model",
                            parts=[types.Part(
                                text="I'm sorry, I can't help with that request. "
                                     "Please contact our support team directly."
                            )]
                        ),
                        finish_reason=types.FinishReason.STOP
                    )
                ]
            )
 
    return None  # Safe to proceed
 
safe_agent = LlmAgent(
    name="safe_customer_support_agent",
    model="gemini-2.5-pro",
    instruction="""You are a customer support agent.
    Answer product questions and provide technical support.
    Never promise refunds or compensation.""",
    before_model_callback=input_guardrail,
)

Why implement Guardrails in callbacks rather than relying solely on the system prompt? System prompts can fail with complex contexts, jailbreak attempts, or ambiguous phrasing. Callbacks provide a deterministic enforcement layer — a last line of defense when the model ignores its instructions.

Output Guardrail: Filter Dangerous Responses After They're Generated

OUTPUT_BLOCKED_PATTERNS = [
    r"refund will be processed|your refund|返金いたします",
    r"you will be compensated|補償します",
    r"\d{4}[-\s]\d{4}[-\s]\d{4}[-\s]\d{4}",  # Credit card pattern
]
 
def output_guardrail(
    callback_context: Any,
    llm_response: types.GenerateContentResponse
) -> Optional[types.GenerateContentResponse]:
    """
    Inspect the model output. Replace dangerous content with a safe response.
    This is set as after_model_callback.
    """
    if not llm_response.candidates:
        return None
 
    candidate = llm_response.candidates[0]
    if not candidate.content or not candidate.content.parts:
        return None
 
    output_text = " ".join(
        part.text for part in candidate.content.parts
        if hasattr(part, "text") and part.text
    )
 
    for pattern in OUTPUT_BLOCKED_PATTERNS:
        if re.search(pattern, output_text, re.IGNORECASE):
            logger.error(
                f"[OUTPUT_GUARDRAIL_TRIGGERED] agent={callback_context.agent_name} "
                f"pattern={pattern} output_preview={output_text[:150]}"
            )
            return types.GenerateContentResponse(
                candidates=[
                    types.Candidate(
                        content=types.Content(
                            role="model",
                            parts=[types.Part(
                                text="For this matter, please contact our customer service team "
                                     "at support@example.com or call 1-800-XXX-XXXX."
                            )]
                        ),
                        finish_reason=types.FinishReason.STOP
                    )
                ]
            )
 
    return None
 
fully_guarded_agent = LlmAgent(
    name="fully_guarded_agent",
    model="gemini-2.5-pro",
    instruction="You are a customer support agent.",
    before_model_callback=input_guardrail,
    after_model_callback=output_guardrail,
)

Token Budget Callback: Preventing Cost Overruns

One of the most common production incidents is token runaway — an agent enters a loop or a user submits an exceptionally long context, burning through thousands of tokens in a single session.

TOKEN_BUDGET_CONFIG = {
    "free_tier": {
        "max_tokens_per_session": 50_000,
        "max_tool_calls_per_session": 20,
    },
    "pro_tier": {
        "max_tokens_per_session": 500_000,
        "max_tool_calls_per_session": 100,
    }
}
 
def token_budget_guardrail(
    callback_context: Any,
    llm_request: types.GenerateContentRequest
) -> Optional[types.GenerateContentResponse]:
    """Enforce per-session token and tool call budgets."""
    state = callback_context.state
    user_tier = state.get("user_tier", "free_tier")
    budget = TOKEN_BUDGET_CONFIG.get(user_tier, TOKEN_BUDGET_CONFIG["free_tier"])
 
    total_tokens = state.get("total_tokens_used", 0)
    tool_calls = state.get("total_tool_calls", 0)
 
    if total_tokens >= budget["max_tokens_per_session"]:
        logger.warning(
            f"[TOKEN_BUDGET_EXCEEDED] session={callback_context.session_id} "
            f"total={total_tokens} limit={budget['max_tokens_per_session']}"
        )
        return types.GenerateContentResponse(
            candidates=[types.Candidate(
                content=types.Content(
                    role="model",
                    parts=[types.Part(
                        text="You've reached the usage limit for this session. "
                             "Please start a new conversation or consider upgrading your plan."
                    )]
                ),
                finish_reason=types.FinishReason.STOP
            )]
        )
 
    if tool_calls >= budget["max_tool_calls_per_session"]:
        return types.GenerateContentResponse(
            candidates=[types.Candidate(
                content=types.Content(
                    role="model",
                    parts=[types.Part(
                        text="You've reached the tool usage limit for this session."
                    )]
                ),
                finish_reason=types.FinishReason.STOP
            )]
        )
 
    return None
 
def token_usage_tracker(
    callback_context: Any,
    llm_response: types.GenerateContentResponse
) -> Optional[types.GenerateContentResponse]:
    """Accumulate token usage in session state after each model call."""
    state = callback_context.state
    usage = llm_response.usage_metadata
 
    if usage:
        current = state.get("total_tokens_used", 0)
        new_tokens = (usage.prompt_token_count or 0) + (usage.candidates_token_count or 0)
        state["total_tokens_used"] = current + new_tokens
 
        logger.info(
            f"[TOKEN_TRACKER] session={callback_context.session_id} "
            f"this_turn={new_tokens} total={state['total_tokens_used']}"
        )
 
    return None

Tool Call Callbacks: Security and Audit Logging

When agents have tools, especially those that make external API calls or modify data, before/after_tool_callbacks become your access control layer.

ALLOWED_API_HOSTS = [
    "api.example.com",
    "data.public-api.org",
]
 
def tool_call_validator(
    tool: Any,
    tool_args: dict,
    tool_context: Any
) -> Optional[dict]:
    """
    Validate tool arguments before execution.
    Return None to allow, return a dict to override the tool result without executing.
    """
    tool_name = tool.name if hasattr(tool, "name") else str(tool)
 
    if tool_name == "make_http_request":
        url = tool_args.get("url", "")
        from urllib.parse import urlparse
        host = urlparse(url).netloc
 
        if not any(host.endswith(allowed) for allowed in ALLOWED_API_HOSTS):
            logger.warning(
                f"[TOOL_BLOCKED] tool={tool_name} blocked_url={url} "
                f"session={tool_context.session_id}"
            )
            return {
                "error": f"Access to {host} is blocked by security policy."
            }
 
    logger.info(
        f"[TOOL_CALL] tool={tool_name} "
        f"args_keys={list(tool_args.keys())} "
        f"session={tool_context.session_id}"
    )
    return None
 
def tool_result_auditor(
    tool: Any,
    tool_args: dict,
    tool_context: Any,
    tool_response: Any
) -> Optional[Any]:
    """Audit tool results for sensitive data after execution."""
    tool_name = tool.name if hasattr(tool, "name") else str(tool)
    result_str = str(tool_response)
 
    logger.info(
        f"[TOOL_RESULT] tool={tool_name} "
        f"result_length={len(result_str)} "
        f"session={tool_context.session_id}"
    )
    return None

The Composite Callback Pattern for Production

In a real system you need logging, guardrails, cost management, and monitoring all at once. Stuffing everything into a single callback function destroys maintainability. The Composite pattern solves this:

from typing import Callable, List, Any, Optional
from google.genai import types
 
def compose_before_model_callbacks(*callbacks: Callable) -> Callable:
    """
    Run multiple before_model_callbacks in sequence.
    Stops and returns the first non-None result (i.e., the first block).
    """
    def composed(callback_context, llm_request):
        for cb in callbacks:
            result = cb(callback_context, llm_request)
            if result is not None:
                return result
        return None
    return composed
 
def compose_after_model_callbacks(*callbacks: Callable) -> Callable:
    """
    Run multiple after_model_callbacks in sequence.
    Each callback receives the (potentially modified) response from the previous one.
    """
    def composed(callback_context, llm_response):
        current_response = llm_response
        override = None
        for cb in callbacks:
            result = cb(callback_context, current_response)
            if result is not None:
                override = result
                current_response = result
        return override
    return composed
 
full_production_agent = LlmAgent(
    name="full_production_agent",
    model="gemini-2.5-pro",
    instruction="""You are a customer support agent.
    Answer questions about our products accurately and helpfully.
    Never promise refunds or compensation.""",
    tools=[product_tool],
    before_model_callback=compose_before_model_callbacks(
        input_guardrail,         # 1. Security filter (highest priority)
        token_budget_guardrail,  # 2. Cost guard
        log_before_model,        # 3. Logging (doesn't block)
        metrics_before_model,    # 4. Metrics recording
    ),
    after_model_callback=compose_after_model_callbacks(
        output_guardrail,        # 1. Output filter (highest priority)
        token_usage_tracker,     # 2. Token accounting
        log_after_model,         # 3. Logging
        metrics_after_model,     # 4. Metrics push
    ),
    before_tool_callback=tool_call_validator,
    after_tool_callback=tool_result_auditor,
)

Each callback has a single responsibility, is independently testable, and can be swapped in or out depending on the environment (dev/staging/production).

Common Mistakes and How to Avoid Them

Mistake 1: Raising exceptions inside callbacks

# ❌ Wrong: crashes the entire agent
def bad_callback(callback_context, llm_request):
    result = fetch_from_database()
    if not result:
        raise ValueError("DB connection failed")  # Agent crash\!
 
# ✅ Correct: catch everything, return None to continue
def good_callback(callback_context, llm_request):
    try:
        result = fetch_from_database()
        if not result:
            logger.warning("DB unavailable, skipping callback")
    except Exception as e:
        logger.error(f"Callback exception (continuing): {e}")
    return None

Callbacks are external code wrapped around your agent. An unhandled exception in a monitoring callback taking down your production service is a textbook "cure worse than the disease" scenario. Always catch exceptions and return None.

Mistake 2: Logging the original response after replacing it

# ❌ The dangerous response ends up in your logs
def bad_guardrail(callback_context, llm_response):
    safe = create_safe_response()
    logger.info(f"Response: {llm_response}")  # Logs the original harmful output
    return safe
 
# ✅ Log that replacement happened, not the full original content
def good_guardrail(callback_context, llm_response):
    safe = create_safe_response()
    logger.warning(
        f"[OUTPUT_REPLACED] original_preview={str(llm_response)[:100]}"
    )
    return safe

Mistake 3: Overly broad patterns causing false positives

Blocking every input containing the word "refund" will catch legitimate inquiries like "How do I request a refund?" Use a monitoring-first rollout strategy:

GUARDRAIL_MODE = os.environ.get("GUARDRAIL_MODE", "monitor")  # monitor → enforce
 
def gradual_guardrail(callback_context, llm_request):
    is_suspicious = check_patterns(llm_request)
    if is_suspicious:
        logger.warning(f"[GUARDRAIL_HIT] mode={GUARDRAIL_MODE}")
        if GUARDRAIL_MODE == "enforce":
            return create_blocked_response()
    return None  # In monitor mode, log only — don't block

Spend 1–2 weeks in monitor mode collecting real traffic data before switching to enforce.

Mistake 4: Thread-safety issues with shared state

# ❌ Global dict shared across concurrent requests — race conditions guaranteed
_token_counts = {}
 
# ✅ Use callback_context.state — ADK manages it per session
def thread_safe_callback(callback_context, llm_request):
    current = callback_context.state.get("token_count", 0)
    callback_context.state["token_count"] = current + estimate_tokens(llm_request)
    return None

Testing Callbacks and Guardrails

Agents with callbacks must be tested with the callbacks included — not just the underlying model behavior.

import pytest
from unittest.mock import MagicMock
from google.genai import types
 
class TestInputGuardrail:
 
    def _make_context(self, agent_name="test_agent", session_id="test-001"):
        ctx = MagicMock()
        ctx.agent_name = agent_name
        ctx.session_id = session_id
        ctx.state = {}
        return ctx
 
    def _make_request(self, text: str):
        return types.GenerateContentRequest(
            contents=[types.Content(
                role="user",
                parts=[types.Part(text=text)]
            )]
        )
 
    def test_blocks_refund_request(self):
        ctx = self._make_context()
        request = self._make_request("I want a refund please")
        result = input_guardrail(ctx, request)
        assert result is not None, "Refund request should be blocked"
        response_text = result.candidates[0].content.parts[0].text
        assert "support team" in response_text.lower()
 
    def test_allows_product_inquiry(self):
        ctx = self._make_context()
        request = self._make_request("How do I set up Product A?")
        result = input_guardrail(ctx, request)
        assert result is None, "Normal product inquiry should pass through"
 
    def test_token_budget_enforced(self):
        ctx = self._make_context()
        ctx.state["total_tokens_used"] = 100_000
        ctx.state["user_tier"] = "free_tier"
        request = self._make_request("Hello")
        result = token_budget_guardrail(ctx, request)
        assert result is not None, "Over-budget sessions should be blocked"
 
# Run with: pytest tests/test_callbacks.py -v

Bringing It All Together

ADK Callbacks and Guardrails transform AI agents from black boxes into transparent, controllable systems. Here's what we implemented:

Input Guardrails intercept harmful requests before the model ever sees them. Output Guardrails catch dangerous model responses before users do. Token Budget Callbacks prevent runaway cost at the session level. Tool Callbacks block unauthorized actions and create a complete audit trail. The Composite Callback pattern keeps each concern in its own function while composing them cleanly.

The recommended rollout path: start by adding only logging callbacks to production. Collect two weeks of real traffic data. Use that data to calibrate your Guardrail patterns, then deploy in monitor mode for another week before switching to enforce. Callbacks can be added and modified without touching core agent logic, so the iteration cycle is fast.