Guardrails & Observability with Callbacks

📦 Companion Code: View the complete, runnable code on GitHub: part11 in the cartpilot-adk2-series repository.

Part 10 hooked into after_model_callback for cost tracking. That's one of six checkpoints ADK gives you across an agent's execution — today we use the rest of them for what they're really for: guardrails, structured audit logging, and policy enforcement that doesn't live inside any single tool's code.

The six hooks

text

before_agent -> before_model -> [LLM call] -> after_model
    -> (if tool call) before_tool -> [tool runs] -> after_tool -> back to before_model
-> after_agent

Each before_* hook can return None (continue normally) or a specific value that short-circuits the step it guards: before_model_callback returning an LlmResponse skips the model call entirely; before_tool_callback returning a dict skips the tool and uses that dict as the result instead. after_* hooks work the same way for the result that already happened — return None to keep it, return a value to replace it.

python

# cartpilot/callbacks.py
import logging
from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.adk.tools.base_tool import BaseTool
from google.adk.tools.tool_context import ToolContext
from google.genai import types

logger = logging.getLogger("cartpilot.audit")
logging.basicConfig(level=logging.INFO)

BLOCKED_TERMS = {"ignore previous instructions", "system prompt"}


def log_request_start(callback_context: CallbackContext):
    """before_agent_callback: structured start-of-turn logging."""
    count = callback_context.state.get("request_count", 0) + 1
    callback_context.state["request_count"] = count
    logger.info("agent_start agent=%s turn=%d", callback_context.agent_name, count)
    return None


def block_prompt_injection(callback_context: CallbackContext, llm_request: LlmRequest):
    """before_model_callback: minimal guardrail example — a real system
    should use a proper classifier here, not a substring blocklist."""
    last_text = ""
    if llm_request.contents:
        last = llm_request.contents[-1]
        if last.parts:
            last_text = (last.parts[0].text or "").lower()

    if any(term in last_text for term in BLOCKED_TERMS):
        logger.warning("blocked_request agent=%s reason=policy", callback_context.agent_name)
        # Returning an LlmResponse skips the model call entirely —
        # zero tokens spent on a request we're not going to honor.
        return LlmResponse(
            content=types.Content(role="model", parts=[types.Part(text="I can't help with that request.")])
        )
    return None


def audit_tool_call(tool: BaseTool, args: dict, tool_context: ToolContext):
    """before_tool_callback: logs every tool invocation before it runs —
    this is the audit trail Part 8's refund approvals need."""
    logger.info("tool_call tool=%s args=%s", tool.name, args)
    return None


def audit_tool_result(tool: BaseTool, args: dict, tool_context: ToolContext, tool_response: dict):
    """after_tool_callback: for sensitive tools like `process_refund`,
    this line IS the audit record of what happened."""
    logger.info("tool_result tool=%s status=%s", tool.name, tool_response.get("status", "unknown"))
    return None


def log_request_end(callback_context: CallbackContext):
    """after_agent_callback: closes out the structured log line for this turn."""
    logger.info("agent_end agent=%s", callback_context.agent_name)
    return None


def attach_standard_callbacks(agent) -> None:
    """Applies the shared guardrail/observability callback set to an agent."""
    agent.before_agent_callback = log_request_start
    agent.before_model_callback = block_prompt_injection
    agent.before_tool_callback = audit_tool_call
    agent.after_tool_callback = audit_tool_result
    agent.after_agent_callback = log_request_end

Applied across CartPilot's specialists:

python

# cartpilot/agent.py (excerpt)
from .callbacks import attach_standard_callbacks
from .cost_tracking import track_token_usage

for _agent in (search_agent, cart_agent, support_agent):
    attach_standard_callbacks(_agent)
    _agent.after_model_callback = track_token_usage

Why the refund audit trail matters specifically here

Go back to Part 8: process_refund pauses for human approval above $50. audit_tool_call and audit_tool_result are what turn "an approval happened" into "a record of who was asked, what they were asked to approve, and what the tool actually did as a result" — the difference between a feature working and a feature you can defend in a compliance review six months later. This is exactly the reasoning behind logging tool calls at the callback layer rather than inside each tool function: you get it once, consistently, for every tool, instead of remembering to add logging code to every sensitive tool individually.

A word on Plugins

ADK's own documentation recommends Plugins over per-agent callbacks specifically for guardrails and policies that should apply globally: a Plugin attaches once at the Runner level instead of being wired onto every agent by hand (notice we had to loop over search_agent, cart_agent, support_agent and remember the coordinator too — a Plugin wouldn't need that loop). We used callbacks here because the mechanics are easier to see laid out per-agent for a tutorial; if you're taking CartPilot's guardrail layer into real production, look at Plugins as the more maintainable home for it.

What's next

Every part of this series has claimed things like "Flash is enough for search" or "the refund flow behaves correctly" — on faith. Part 12 replaces faith with an actual evaluation suite: a way to check, automatically and repeatably, whether CartPilot still does the right thing after you change a prompt, swap a model, or refactor a tool.