ADK Callbacks (6 types)

ADK Callbacks (6 types)

Original Notion page had a diagram of the 6 callback hooks. See migrated_from URL.

ADK exposes 6 lifecycle hooks across three levels — agent, model, and tool — each with a before_* and after_* variant.

The 6 callbacks

Before Agent Callback

Triggered before anything is called in the agentic system. Use it to set up resources and state before a specific agent runs (“state hydration”). E.g., grab the user’s order history.

After Agent Callback

Triggered after the agent finishes. Clean-up tasks, post-execution validation, logging.

Before Model Callback

Fires before we send the request to the model (OpenAI, Claude, Gemini, etc.). Useful for:

• Adding dynamic instructions
• Implementing guardrails (filter requests)
• Quitting the loop / skipping if the user asks something disallowed

After Model Callback

Alter the response coming back from the model. Reformat responses, censor information, log outputs.

Before Tool Callback

Allows inspection and modification of tool arguments. Perform authorization checks.

After Tool Callback

Allows inspection and modification of tool results. Perform authorization checks, log the tool result, save to state.

Notes

• All callbacks can pull from and write to state via callback_context.state.
• Returning None continues normal execution. Returning a non-None value typically short-circuits the step (skip the model, override the response, etc.).

---

Example 1 — Before/After Agent (log request duration)

# Create the Agent
root_agent = LlmAgent(
    name="before_after_agent",
    model="gemini-2.0-flash",
    description="A basic agent that demonstrates before and after agent callbacks",
    instruction="""
    You are a friendly greeting agent. Your name is {agent_name}.
 
    Your job is to:
    - Greet users politely
    - Respond to basic questions
    - Keep your responses friendly and concise
    """,
    before_agent_callback=before_agent_callback,
    after_agent_callback=after_agent_callback,
)

def before_agent_callback(callback_context: CallbackContext) -> Optional[types.Content]:
    """
    Simple callback that logs when the agent starts processing a request.
 
    Returns:
        None to continue with normal agent processing
    """
    state = callback_context.state
    timestamp = datetime.now()
 
    # Set agent name if not present
    if "agent_name" not in state:
        state["agent_name"] = "SimpleChatBot"
 
    # Initialize request counter
    if "request_counter" not in state:
        state["request_counter"] = 1
    else:
        state["request_counter"] += 1
 
    # Store start time for duration calculation in after_agent_callback
    state["request_start_time"] = timestamp
 
    print("=== AGENT EXECUTION STARTED ===")
    print(f"Request #: {state['request_counter']}")
    print(f"Timestamp: {timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"\n[BEFORE CALLBACK] Agent processing request #{state['request_counter']}")
 
    return None

def after_agent_callback(callback_context: CallbackContext) -> Optional[types.Content]:
    """
    Simple callback that logs when the agent finishes processing a request.
    # Return None if all's good
    # Return Content if you want to override what's returned to the user
    """
    state = callback_context.state
 
    timestamp = datetime.now()
    duration = None
    if "request_start_time" in state:
        duration = (timestamp - state["request_start_time"]).total_seconds()
 
    print("=== AGENT EXECUTION COMPLETED ===")
    print(f"Request #: {state.get('request_counter', 'Unknown')}")
    if duration is not None:
        print(f"Duration: {duration:.2f} seconds")
    print(
        f"[AFTER CALLBACK] Agent completed request #{state.get('request_counter', 'Unknown')}"
    )
    if duration is not None:
        print(f"[AFTER CALLBACK] Processing took {duration:.2f} seconds")
 
    return None

Example 2 — Before/After Model (content filter)

root_agent = LlmAgent(
    name="content_filter_agent",
    model="gemini-2.0-flash",
    description="An agent that demonstrates model callbacks for content filtering and logging",
    instruction="""
    You are a helpful assistant.
 
    Your job is to:
    - Answer user questions concisely
    - Provide factual information
    - Be friendly and respectful
    """,
    before_model_callback=before_model_callback,
    after_model_callback=after_model_callback,
)

def before_model_callback(
    callback_context: CallbackContext,
    llm_request: LlmRequest,                 # include the LLM request
) -> Optional[LlmResponse]:
    """
    Filters inappropriate content and logs request info.
    Returning a non-None LlmResponse will SKIP the model call.
    """
    state = callback_context.state
    agent_name = callback_context.agent_name
 
    # Extract the last user message
    last_user_message = ""
    if llm_request.contents and len(llm_request.contents) > 0:
        for content in reversed(llm_request.contents):
            if content.role == "user" and content.parts and len(content.parts) > 0:
                if hasattr(content.parts[0], "text") and content.parts[0].text:
                    last_user_message = content.parts[0].text
                    break
 
    print("=== MODEL REQUEST STARTED ===")
    print(f"Agent: {agent_name}")
    if last_user_message:
        print(f"User message: {last_user_message[:100]}...")
        state["last_user_message"] = last_user_message
    else:
        print("User message: <empty>")
    print(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
 
    # Check for inappropriate content
    if last_user_message and "sucks" in last_user_message.lower():
        print("=== INAPPROPRIATE CONTENT BLOCKED ===")
        print("[BEFORE MODEL] ⚠️ Request blocked due to inappropriate content")
 
        # Return a response to skip the model call
        return LlmResponse(
            content=types.Content(
                role="model",
                parts=[
                    types.Part(
                        text="I cannot respond to messages containing inappropriate language. "
                        "Please rephrase your request without using words like 'sucks'."
                    )
                ],
            )
        )
 
    state["model_start_time"] = datetime.now()
    print("[BEFORE MODEL] ✓ Request approved for processing")
 
    # Return None to proceed with normal model request
    return None

def after_model_callback(
    callback_context: CallbackContext,
    llm_response: LlmResponse,
) -> Optional[LlmResponse]:
    """
    Replaces negative words with more positive alternatives.
    """
    print("[AFTER MODEL] Processing response")
 
    if not llm_response or not llm_response.content or not llm_response.content.parts:
        return None
 
    response_text = ""
    for part in llm_response.content.parts:
        if hasattr(part, "text") and part.text:
            response_text += part.text
 
    if not response_text:
        return None
 
    replacements = {
        "problem": "challenge",
        "difficult": "complex",
    }
 
    modified_text = response_text
    modified = False
    for original, replacement in replacements.items():
        if original in modified_text.lower():
            modified_text = modified_text.replace(original, replacement)
            modified_text = modified_text.replace(
                original.capitalize(), replacement.capitalize()
            )
            modified = True
 
    if modified:
        print("[AFTER MODEL] ↺ Modified response text")
        modified_parts = [copy.deepcopy(part) for part in llm_response.content.parts]
        for i, part in enumerate(modified_parts):
            if hasattr(part, "text") and part.text:
                modified_parts[i].text = modified_text
        return LlmResponse(content=types.Content(role="model", parts=modified_parts))
 
    return None

Example 3 — Before/After Tool

root_agent = LlmAgent(
    name="tool_callback_agent",
    model="gemini-2.0-flash",
    description="An agent that demonstrates tool callbacks by looking up capital cities",
    instruction="""
    You are a helpful geography assistant.
 
    Your job is to:
    - Find capital cities when asked using the get_capital_city tool
    - Use the exact country name provided by the user
    - ALWAYS return the EXACT result from the tool, without changing it
    - When reporting a capital, display it EXACTLY as returned by the tool
 
    Examples:
    - "What is the capital of France?" → Use get_capital_city with country="France"
    - "Tell me the capital city of Japan" → Use get_capital_city with country="Japan"
    """,
    tools=[get_capital_city],
    before_tool_callback=before_tool_callback,
    after_tool_callback=after_tool_callback,
)

def before_tool_callback(
    tool: BaseTool, args: Dict[str, Any], tool_context: ToolContext
) -> Optional[Dict]:
    """
    Modifies tool arguments or skips the tool call.
    """
    tool_name = tool.name
    print(f"[Callback] Before tool call for '{tool_name}'")
    print(f"[Callback] Original args: {args}")
 
    # If someone asks about 'Merica, convert to United States
    if tool_name == "get_capital_city" and args.get("country", "").lower() == "merica":
        print("[Callback] Converting 'Merica to 'United States'")
        args["country"] = "United States"
        print(f"[Callback] Modified args: {args}")
        return None   # means continue
 
    # Skip the call completely for restricted countries
    if (
        tool_name == "get_capital_city"
        and args.get("country", "").lower() == "restricted"
    ):
        print("[Callback] Blocking restricted country")
        return {"result": "Access to this information has been restricted."}
 
    print("[Callback] Proceeding with normal tool call")
    return None

def after_tool_callback(
    tool: BaseTool, args: Dict[str, Any], tool_context: ToolContext, tool_response: Dict
) -> Optional[Dict]:
    """
    Modifies the tool response after execution.
    """
    tool_name = tool.name
    print(f"[Callback] After tool call for '{tool_name}'")
    print(f"[Callback] Args used: {args}")
    print(f"[Callback] Original response: {tool_response}")
 
    original_result = tool_response.get("result", "")
    print(f"[Callback] Extracted result: '{original_result}'")
 
    # Add a note for any USA capital responses
    if tool_name == "get_capital_city" and "washington" in original_result.lower():
        print("[Callback] DETECTED USA CAPITAL - adding patriotic note!")
        modified_response = copy.deepcopy(tool_response)
        modified_response["result"] = (
            f"{original_result} (Note: This is the capital of the USA. 🇺🇸)"
        )
        modified_response["note_added_by_callback"] = True
        print(f"[Callback] Modified response: {modified_response}")
        return modified_response
 
    print("[Callback] No modifications needed, returning original response")
    return None

See next

• ADK-Sessions-and-State — callbacks read/write the same state
• ADK-Tools — tools are the layer below before_tool_callback