google-adk-callbacks

name	google-adk-callbacks
description	Implement ADK agent and tool callbacks. Use when adding lifecycle hooks — before/after agent execution, before/after tool calls, for logging, validation, conditional execution, or response modification.

Google ADK — Callbacks

Types of Callbacks

Callback	Signature	Purpose
before_agent_callback	(CallbackContext) -> Optional[Content]	Gate or modify before agent runs
after_agent_callback	(CallbackContext) -> Optional[Content]	Post-process after agent completes
before_tool_callback	(tool: BaseTool, args: dict, tool_context: ToolContext) -> Optional[dict]	Validate/modify tool inputs
after_tool_callback	(tool: BaseTool, args: dict, tool_context: ToolContext, tool_response: dict) -> Optional[dict]	Transform tool outputs

Imports

from google.adk.agents import Agent
from google.adk.agents.callback_context import CallbackContext
from google.adk.tools.tool_context import ToolContext
from google.adk.tools.base_tool import BaseTool
from google.genai import types
from typing import Optional

before_agent_callback

Return None to proceed normally, or Content to skip the agent and return that content instead.

def check_authorization(callback_context: CallbackContext) -> Optional[types.Content]:
    """Only allow the agent to run if user is authorized."""
    if not callback_context.state.get("is_authorized"):
        return types.Content(
            parts=[types.Part(text="You must be authorized to use this agent.")]
        )
    return None  # Proceed normally

agent = Agent(
    name="secure_agent",
    model="gemini-2.5-flash",
    instruction="Handle sensitive operations.",
    before_agent_callback=check_authorization,
)

after_agent_callback

def log_completion(callback_context: CallbackContext) -> Optional[types.Content]:
    """Log when agent completes and optionally override response."""
    callback_context.state["last_agent_completed"] = callback_context.agent_name
    return None  # Don't override response

agent = Agent(
    name="logged_agent",
    model="gemini-2.5-flash",
    instruction="...",
    after_agent_callback=log_completion,
)

before_tool_callback

Return None to proceed, or a dict to skip the tool and use the dict as the result.

def validate_tool_input(
    tool: BaseTool,
    args: dict,
    tool_context: ToolContext,
) -> Optional[dict]:
    """Validate and potentially block tool calls."""
    if tool.name == "delete_file" and not tool_context.state.get("admin"):
        return {"error": "Only admins can delete files."}
    return None  # Proceed with tool execution

agent = Agent(
    name="guarded_agent",
    model="gemini-2.5-flash",
    instruction="Manage files for users.",
    tools=[delete_file, read_file],
    before_tool_callback=validate_tool_input,
)

after_tool_callback

def redact_sensitive_data(
    tool: BaseTool,
    args: dict,
    tool_context: ToolContext,
    tool_response: dict,
) -> Optional[dict]:
    """Redact PII from tool results before returning to LLM."""
    if tool.name == "get_user_profile":
        if "ssn" in tool_response:
            tool_response["ssn"] = "***-**-****"
    return tool_response  # Return modified result

agent = Agent(
    name="privacy_agent",
    model="gemini-2.5-flash",
    tools=[get_user_profile],
    after_tool_callback=redact_sensitive_data,
)

Conditional Agent Execution (Factory Pattern)

from google.adk.agents.base_agent import BeforeAgentCallback

def skip_if_not_relevant(agent_name: str) -> BeforeAgentCallback:
    """Factory: creates a callback that skips agent if not in execution plan."""
    def callback(callback_context: CallbackContext) -> Optional[types.Content]:
        execution_agents = callback_context.state.get("execution_agents", [])
        if agent_name not in execution_agents:
            return types.Content(
                parts=[types.Part(text=f"Skipping {agent_name} — not relevant.")]
            )
        return None
    return callback

code_agent = Agent(
    name="code_agent",
    model="gemini-2.5-flash",
    instruction="Generate code.",
    before_agent_callback=skip_if_not_relevant("code_agent"),
)

CallbackContext Properties

CallbackContext (alias for Context) provides access to agent state:

Property	Type	Description
callback_context.state	State	Read/write session state

For tool callbacks, use tool_context.state instead (ToolContext is also an alias for Context).

Common Patterns

Rate Limiting

import time

def rate_limit(callback_context: CallbackContext) -> Optional[types.Content]:
    last_call = callback_context.state.get("last_tool_call_time", 0)
    if time.time() - last_call < 1.0:
        return types.Content(parts=[types.Part(text="Rate limited. Try again.")])
    callback_context.state["last_tool_call_time"] = time.time()
    return None

Audit Logging

def audit_log(
    tool: BaseTool,
    args: dict,
    tool_context: ToolContext,
) -> Optional[dict]:
    logs = tool_context.state.get("audit_log", [])
    logs.append({"tool": tool.name, "args": args, "timestamp": time.time()})
    tool_context.state["audit_log"] = logs
    return None  # Don't modify execution

Key Rules

• Return None to proceed normally
• Return Content (before/after_agent) or dict (before/after_tool) to override
• Callbacks can be sync or async
• Tool callbacks receive (tool, args, tool_context) — use tool_context.state for state
• Agent callbacks receive (callback_context) — use callback_context.state for state
• Factory pattern (returning a callback from a function) enables parameterized callbacks

Related Skills

• google-adk-llm-agent — Agent configuration (where callbacks are attached)
• google-adk-function-tool — Tool creation (tools that callbacks intercept)