What are Callbacks and Plugins in Google ADK? Explain the 6 callback hooks and plugin system.

Callbacks & Plugins in Google ADK

Google ADK provides 6 callback hooks and a plugin system for intercepting and modifying agent behavior at every stage of execution.

The 6 Callback Hooks

before_agent_callback

after_agent_callback

before_model_callback

after_model_callback

before_tool_callback

after_tool_callback

Using Callbacks

Logging Callback

pythonCopy
from google.adk.agents import Agent

def log_before_model(callback_context, llm_request):
    messages = llm_request.contents
    print(f"[MODEL] Calling with {len(messages)} messages")
    return None   # return None to proceed normally

def log_after_model(callback_context, llm_response):
    text = llm_response.candidates[0].content.parts[0].text
    print(f"[MODEL] Response: {text[:100]}...")
    return None

def log_before_tool(callback_context, tool_call):
    print(f"[TOOL] Calling: {tool_call.function_call.name}")
    return None

def log_after_tool(callback_context, tool_response):
    print(f"[TOOL] Result: {str(tool_response)[:100]}")
    return None

agent = Agent(
    name="logged_agent",
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant.",
    tools=[get_weather],
    before_model_callback=log_before_model,
    after_model_callback=log_after_model,
    before_tool_callback=log_before_tool,
    after_tool_callback=log_after_tool,
)

Guardrail Callback (Block Unsafe Queries)

pythonCopy
from google.genai import types

BLOCKED_WORDS = ["hack", "exploit", "jailbreak"]

def safety_guardrail(callback_context, llm_request):
    user_msg = llm_request.contents[-1].parts[0].text.lower()

    if any(word in user_msg for word in BLOCKED_WORDS):
        # Return a response to skip the LLM call entirely
        return types.Content(
            role="model",
            parts=[types.Part(text="I cannot help with that request.")]
        )
    return None   # proceed normally

agent = Agent(
    name="safe_agent",
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant.",
    before_model_callback=safety_guardrail,
)

Caching Callback

pythonCopy
import hashlib

_cache = {}

def cache_before_model(callback_context, llm_request):
    key = hashlib.md5(str(llm_request.contents).encode()).hexdigest()
    if key in _cache:
        print("[CACHE] Hit!")
        return _cache[key]
    callback_context.state["temp:cache_key"] = key
    return None

def cache_after_model(callback_context, llm_response):
    key = callback_context.state.get("temp:cache_key")
    if key:
        _cache[key] = llm_response
    return None

Plugin System

Plugins are global-scope callbacks registered on the Runner (not individual agents). They apply to ALL agents in the system.

pythonCopy
from google.adk.runners import Runner

class AuditPlugin:
    """Audit all agent and tool calls."""

    def before_agent(self, callback_context, *args):
        print(f"[AUDIT] Agent {callback_context.agent_name} starting")

    def after_agent(self, callback_context, *args):
        print(f"[AUDIT] Agent {callback_context.agent_name} completed")

    def before_tool(self, callback_context, tool_call, *args):
        print(f"[AUDIT] Tool call: {tool_call.function_call.name}")

class CostTrackerPlugin:
    """Track token usage across all agents."""

    def __init__(self):
        self.total_tokens = 0

    def after_model(self, callback_context, llm_response, *args):
        tokens = llm_response.usage_metadata.total_token_count
        self.total_tokens += tokens
        print(f"[COST] Total tokens: {self.total_tokens}")

# Register plugins on the Runner
runner = Runner(
    agent=root_agent,
    app_name="my_app",
    session_service=InMemorySessionService(),
    plugins=[AuditPlugin(), CostTrackerPlugin()],
)

Execution Order

Rule: Plugin callbacks run BEFORE Agent callbacks. If a Plugin returns a non-empty response, the Agent-level callback is skipped.

Pre-built Plugins

Learn more at Callbacks and Plugins.