Give an agent typed function tools

When you need this. Our docs assistant can read a page. Now a developer wants it to file a documentation issue. Reading is one operation; writing requires valid arguments and a person’s approval. We will add those two checks before a proposed issue reaches the local tracker.

The tools search and read the bundled docuctl documentation. file_docs_issue appends to an in-memory ISSUES list so you can inspect writes without connecting a real issue service.

Write the tools as typed functions

Download the lab using Download code and open adk-function-tools-lesson. Create the environment and configure your private Gemini key in .env:

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

The Python entrypoint loads this root .env. We will use the tokens page as the input to the write exercise:

# Authenticate docuctl in CI

CI jobs authenticate with a deploy token, not a personal login. Create one with:

    docuctl tokens create --name ci-docs-api --scope deploy:staging

The token is shown once. Store it in your CI secret store as `DOCUCTL_TOKEN`. Tokens expire after 90 days; `docuctl tokens list` shows each token's expiry.

The tool definitions separate read operations from the write:

"""Typed function tools for the docs assistant: two reads and one write."""
from pathlib import Path
from typing import Annotated, Literal

from pydantic import Field

DOCS = Path(__file__).resolve().parent / "docs"
Page = Literal["config", "deploy", "rollback", "tokens"]
ISSUES: list[dict] = []  # every docs issue the tool filed, in order


def search_docs(query: Annotated[str, Field(min_length=2)], limit: Annotated[int, Field(ge=1, le=5)] = 3) -> dict:
    """Find documentation pages whose text contains every word in the query.

    Args:
        query: Words to look for, for example "rollback production".
        limit: The largest number of pages to return, from 1 to 5.
    """
    words = query.lower().split()
    hits = []
    for path in sorted(DOCS.glob("*.md")):
        text = path.read_text(encoding="utf-8")
        if all(word in text.lower() for word in words):
            hits.append({"page": path.stem, "title": text.splitlines()[0].lstrip("# ")})
    return {"pages": hits[:limit], "total": len(hits)}


def read_page(page: Page) -> dict:
    """Read one documentation page in full."""
    return {"page": page, "text": (DOCS / f"{page}.md").read_text(encoding="utf-8")}


def file_docs_issue(
    page: Page,
    kind: Literal["outdated", "incorrect", "unclear"],
    summary: Annotated[str, Field(min_length=10, max_length=200)],
) -> dict:
    """File an issue against a documentation page. This writes to the docs issue tracker.

    Args:
        page: The page the issue is about.
        kind: outdated, incorrect or unclear.
        summary: One sentence that says what is wrong, 10 to 200 characters.
    """
    issue = {"id": f"DOC-{len(ISSUES) + 1}", "page": page, "kind": kind, "summary": summary}
    ISSUES.append(issue)
    return issue

Literal restricts page names and issue categories. Annotated attaches Field constraints: a search limit from 1 to 5, and a summary from 10 to 200 characters. The docstrings explain each tool to the model. These declarations help it propose a call; application code must still validate that proposal.

Validate calls and require approval for the write

The agent uses a callback to validate arguments, then a FunctionTool confirmation gate for the write:

@functools.cache
def arguments_model(func) -> type[BaseModel]:
    """A Pydantic model of the function's parameters, Field constraints included."""
    hints = typing.get_type_hints(func, include_extras=True)
    fields = {name: (hints[name], ... if p.default is p.empty else p.default)
              for name, p in inspect.signature(func).parameters.items()}
    return create_model(f"{func.__name__}_args", __config__=ConfigDict(extra="forbid"), **fields)


def check_arguments(tool, args: dict, tool_context) -> dict | None:
    """Validate a call before it runs or asks for approval. A dict result replaces the call."""
    func = getattr(tool, "func", None)
    if func is None:
        return None
    try:
        arguments_model(func).model_validate(args)
    except ValidationError as error:
        problems = [f"{'.'.join(map(str, e['loc']))}: {e['msg']}" for e in error.errors()]
        return {"error": "Invalid arguments. " + "; ".join(problems)}
    return None  # None lets the call continue


def build_agent(model=None) -> Agent:
    return Agent(
        name="docs_assistant",
        model=model or (LiteLlm(model=MODEL) if MODEL.startswith("ollama_chat/") else MODEL),
        instruction=INSTRUCTION,
        before_tool_callback=check_arguments,
        tools=[
            docs_tools.search_docs,
            docs_tools.read_page,
            FunctionTool(docs_tools.file_docs_issue, require_confirmation=True),
        ],
        generate_content_config=types.GenerateContentConfig(
            temperature=0.2, max_output_tokens=1024,
            thinking_config=types.ThinkingConfig(thinking_level="low")),
    )

arguments_model() keeps the Annotated metadata when constructing a Pydantic model. extra="forbid" rejects unexpected fields. Returning an error dictionary from check_arguments replaces the tool result; returning None lets the call continue. Validation therefore happens before approval as well as before execution.

Plain read functions are wrapped by ADK automatically. The write is wrapped explicitly to set require_confirmation=True. All model turns request LOW thinking and at most 1,024 output tokens.

A Field limit that ADK does not check. In this SDK, declaring Field(min_length=10) alone does not enforce that bound at the tool boundary. Without the validation callback, a short summary can reach confirmation. Keep the Pydantic check in the execution path.

An unknown or missing argument. The callback rejects missing fields and extra fields and returns the validation problem to the model. The tool does not run.

The application also owns the approval interaction:

async def ask(runner: Runner, session_id: str, text: str, approve: Approver) -> list:
    """Run one user turn. When a write asks for confirmation, let `approve` decide and resume."""
    events = []
    message = types.Content(role="user", parts=[types.Part(text=text)])
    while message:
        pending = []
        async for event in runner.run_async(user_id="dev", session_id=session_id, new_message=message):
            events.append(event)
            pending += [c for c in event.get_function_calls() if c.name == CONFIRM]
        replies = []
        for request in pending:
            call = request.args["originalFunctionCall"]
            replies.append(types.Part(function_response=types.FunctionResponse(
                id=request.id, name=CONFIRM, response={"confirmed": approve(call["name"], call["args"])})))
        message = types.Content(role="user", parts=replies) if replies else None
    return events

ADK emits an adk_request_confirmation call with the original tool name and arguments. The application asks a person, then sends a FunctionResponse with the same call ID and confirmed set to true or false. That response resumes the invocation.

A rejected call retried by the model. A model can propose another write after receiving a rejection. Every new call still needs its own approval; the agent’s system instruction asks it to stop. For a hard application rule, track the rejection and refuse further writes rather than relying only on that instruction.

Run the scenarios

From the lab directory, request a specific issue:

python agent.py "Read the tokens page, then file an unclear issue with the summary: Token expiry details are unclear."

The terminal asks Allow file_docs_issue({...})? [y/N]. Read the page, category and summary in that prompt. Enter n first. Run the command again and enter y only if the proposed issue matches your request. Each command starts a new process and a new local issue list.

A declined write. The confirmation response prevents the write, and ISSUES stays empty. Check that the final answer also says nothing was filed. Approval should be about the proposed operation, not the fluency of the answer.

The saved confirmation cases illustrate the difference:

Recorded example. The archived “Give an agent typed function tools” demonstration retains the earlier shipctl scenario; the current lab uses docuctl and does not claim a new model result.

approved: approvals=1, issues_written=1
declined: approvals=1, issues_written=0

For a read-only turn, try:

python agent.py "Find the rollback page and summarize it. Do not file an issue."

Inspect the printed call and result lines. A read should not trigger the write-approval prompt. ISSUES is a teaching fixture and disappears at process exit; connecting a real tracker also requires authentication, authorization and protection against duplicate writes.

From function schema to tool response

The model sees tool names, descriptions and parameter schemas. It proposes a function call; ADK runs the validation callback, asks for confirmation when required, invokes the function and returns its result to the model. The model then explains the result to the developer.

A schema influences a proposal. Validation enforces the argument contract. Confirmation controls whether an otherwise valid write may proceed. None of these checks establishes that an issue’s wording is factually correct; review it against the documentation before approving.

Try a validation failure

Ask for an issue with the exact summary expiry?. Inspect whether the model corrects the summary before calling the tool or receives a validation error. In both cases, the invalid summary must not reach the write. The offline tests deliberately submit invalid arguments so that this boundary is checked independently of model behavior.

Test the tools without a model

python -m unittest -v test_agent.py test_cleanup.py

The tests replay saved model turns through the actual callback, tools and approval loop. They check invalid arguments, accepted and declined writes, and repeated confirmation requests without a provider call.

Next, connect the assistant to an MCP server so it can use tools maintained in another process.

References

• ADK function tools: typed functions and tool declarations.
• ADK tool confirmation: confirmation requests and responses.
• ADK callbacks: intercepting a tool call.
• Pydantic fields: constraints and metadata.