Guard model and tool calls with callbacks and plugins

When you need this. A developer asks the assistant to check ship.toml, but the file contains a credential identifier. Another request asks it to read a path outside the project. We need code that removes the matching identifier before storage or model use, and code that refuses an out-of-workspace read before the tool runs.

An agent callback guards one agent’s operation. An App plugin applies policy across the application. We will use a plugin for message and tool-result redaction, and a before_tool_callback for the file tool’s path check.

Look at the workspace

Download the lab using Download code and open adk-callbacks-plugins-lesson. Configure your Gemini key in the private root .env:

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
cp .env.example .env

The entrypoint loads this .env. Use only the supplied example identifier when testing redaction; do not paste a real credential. The input file is:

[service]
name = "docs-api"
build = "make build"

[env.staging]
LOG_LEVEL = "debug"
AWS_ACCESS_KEY_ID = "AKIAIOSFODNN7EXAMPLE"

The tool should return useful configuration text with a marker in place of the matching access-key ID. The example detects one identifier format; it is not a general secret scanner.

Write the guards

RedactSecrets handles two entry points: user text before the Runner stores it, and tool text before it reaches the model or stored tool result:

AWS_KEY_ID = re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b")  # AWS access key IDs
REDACTED = "[REDACTED aws_access_key_id]"


def redact(text: str) -> tuple[str, int]:
    return AWS_KEY_ID.subn(REDACTED, text)


class RedactSecrets(BasePlugin):
    """App-wide: remove AWS key IDs from user messages and tool results before they are stored or sent."""

    def __init__(self) -> None:
        super().__init__(name="redact_secrets")
        self.redactions: list[str] = []  # where each redaction happened, for an audit log

    async def on_user_message_callback(self, *, invocation_context, user_message):
        changed = False
        parts = []
        for part in user_message.parts or []:
            if part.text:
                text, count = redact(part.text)
                changed |= count > 0
                self.redactions += ["user_message"] * count
                part = part.model_copy(update={"text": text})
            parts.append(part)
        return types.Content(role=user_message.role, parts=parts) if changed else None  # None keeps it

    async def after_tool_callback(self, *, tool, tool_args, tool_context, result):
        if isinstance(result, dict) and isinstance(result.get("text"), str):
            text, count = redact(result["text"])
            if count:
                self.redactions += [f"tool:{tool.name}"] * count
                return {**result, "text": text}  # a dict replaces the tool result
        return None

A callback returning None leaves the value unchanged. Returning a replacement message or result changes what the runtime passes onward. The plugin logs whether each redaction occurred in a user message or a named tool result, so the audit log does not copy the secret.

Next, check a proposed file path before opening it:

def read_file(path: str) -> dict:
    """Read a text file from the repository workspace, for example ship.toml or docs/rollback.md."""
    target = WORKSPACE / path
    if not target.is_file():
        return {"error": f"No file {path}."}
    return {"path": path, "text": target.read_text(encoding="utf-8")}


def stay_in_workspace(tool, args: dict, tool_context) -> dict | None:
    """Agent callback: refuse a path that resolves outside the workspace, before the tool runs."""
    if tool.name != "read_file":
        return None
    resolved = (WORKSPACE / str(args.get("path", ""))).resolve()
    if not resolved.is_relative_to(WORKSPACE):  # a string prefix check would accept workspace-old/
        return {"error": f"Refused: {args.get('path')} is outside the workspace."}
    return None

resolve() normalizes the path, and is_relative_to() checks its relationship to the workspace. A string-prefix check would confuse workspace/ with workspace-old/. Returning an error dictionary prevents the file read.

The agent and plugin are assembled into one App:

root_agent = Agent(
    name="docs_assistant",
    model=LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL,
    instruction="""You help developers with docuctl and their repository. Read files with read_file;
the docuctl docs are in docs/. Check a config file against docs/config.md. Keep answers under 100 words.""",
    tools=[read_file],
    before_tool_callback=stay_in_workspace,
    generate_content_config=types.GenerateContentConfig(
        temperature=0.2, max_output_tokens=1024, thinking_config=types.ThinkingConfig(thinking_level="low")),
)


def build_app(agent: Agent = root_agent, plugins: list[BasePlugin] | None = None, *, app_name: str = "docs") -> App:
    """Plugins on the App apply to every agent, tool and model call the runner makes."""
    return App(name=app_name, root_agent=agent, plugins=[RedactSecrets()] if plugins is None else plugins)

The Runner receives that App, not a bare agent, so the plugin stays in the execution path. The model uses LOW thinking and at most 1,024 output tokens per response.

Run the guarded assistant

Ask it to inspect the supplied file:

python guarded.py "Read ship.toml and compare it with docs/config.md. What should change?"

Inspect the read_file calls and final answer. The assistant should still identify the configuration problem when the matching identifier is replaced with [REDACTED aws_access_key_id].

Now try a refused path:

python guarded.py "Read ../workspace-old/ship.toml and compare it with ship.toml."

Look for the tool’s Refused result. A model saying it will not read the file is different from the callback preventing the read. The offline tests check the callback boundary directly.

For input-message redaction, paste only the supplied example ID AKIAIOSFODNN7EXAMPLE into a test question. The plugin must remove it before the Runner appends that user message to the session.

A small comparison from the supplied saved cases shows why both request and storage checks matter:

pasted: key_sent_to_model=false, key_in_session=false
pasted-no-plugin: key_sent_to_model=true, key_in_session=true

Intercept model and tool requests

The critical ordering is redact the user message, append it to the session, then call the model. before_model_callback runs after the append, so redacting only there can protect the model request while leaving the original text in session storage. on_user_message_callback runs early enough to change both.

For tool calls, plugin callbacks run before the agent’s callbacks. A plugin’s before_tool_callback that returns a value short-circuits the remaining callbacks and the tool; otherwise the agent’s callback gets its turn. After-tool callbacks use the same first-replacement rule. Keep plugin order deliberate, especially when a logging plugin could observe text before redaction.

When a guard is in the wrong place

No redaction. The original identifier can reach both the model request and the stored user event. Register the plugin on the App that the Runner actually uses.

Redaction only before the model call. The model sees redacted text, but the earlier stored event can still contain the identifier. Redact the incoming user message before it is appended.

A key in a tool result. Cleaning user messages alone misses identifiers read from files. Redact tool results as well.

A path outside the workspace. The resolved-path callback returns an error before read_file runs. This is a rule for one tool, not operating-system isolation. If another process can change files between the check and the read, it can introduce a symlink race; use an isolated filesystem or a safe descriptor-based open for that threat model.

A plugin that answers for a tool. A plugin that supplies a result prevents later agent callbacks from running. Put application-wide refusal rules early in the plugin chain and avoid answering for tools the plugin does not own.

Test the guards without a model

python -m unittest -v test_guarded.py

The tests replay saved model turns through the real App, plugin, callbacks and session service. They inspect both outgoing requests and stored events, including the deliberately incorrect model-only redaction strategy. No provider call is needed.

Next, build a graph workflow to make the order of multiple processing steps explicit.

References

• ADK callbacks: hook inputs, replacements and execution order.
• ADK plugins: application-wide callbacks.
• ADK Apps: runtime configuration and plugin registration.
• AWS access keys: identifiers and example values.