Custom tools

Overview

When the built-in tools don’t fit your needs, wrap any Python function as a custom tool to extend your agent. This page covers three forms: plain-argument functions, context-aware functions, and long-running tasks.

A plain-argument function is the simplest custom tool: define a function, add type hints and a docstring, and register it on the agent. The agent decides when to call it and how to pass arguments based on the signature and docstring.
A context-aware function adds a tool_context: ToolContext parameter to access the agent’s runtime context — shared session state and other runtime information. The framework injects this argument automatically; the model neither sees nor needs to pass it.
Long-running tasks suit time-consuming or asynchronous work (large computations, data analysis, batch jobs). Wrap the function with LongRunningFunctionTool: the tool returns a pending status and a task ID first, your app advances the task in the background, then feeds the final result back to the agent.

Prerequisites

Complete model configuration before running the agent examples. Plain Python functions can also be called locally without a model or service credentials.

Usage

Plain-argument functions

Define the function

Use flat, clear arguments and return types.

Write the docstring

Describe what the function does, its arguments, and its return value — the model relies on this to understand the tool.

Put the function in the agent’s tools list.

The calculator below illustrates this. Note the divide branch: on division by zero it returns status: "error" with no result, consistent with the status: "success" of the other branches.

examples/tools/function_tools/simple_function_tool.py

import asyncio
from typing import Any, Dict

from veadk import Agent, Runner


def calculator(
    a: float, b: float, operation: str
) -> Dict[str, Any]:
    """A simple calculator that performs a basic arithmetic operation.

    Args:
        a (float): The first operand.
        b (float): The second operand.
        operation (str): One of "add", "subtract", "multiply", "divide".

    Returns:
        Dict[str, Any]: On success, contains "result", "operation", and
        "status" == "success". On failure (unsupported operation or division
        by zero), contains "status" == "error" and a "message".
    """
    if operation == "add":
        return {"result": a + b, "operation": "+", "status": "success"}
    if operation == "subtract":
        return {"result": a - b, "operation": "-", "status": "success"}
    if operation == "multiply":
        return {"result": a * b, "operation": "*", "status": "success"}
    if operation == "divide":
        if b == 0:
            return {"status": "error", "message": "Divisor cannot be zero."}
        return {"result": a / b, "operation": "/", "status": "success"}
    return {"status": "error", "message": f"Unsupported operation: {operation}"}


agent = Agent(
    name="computing_agent",
    model_name="doubao-seed-2-1-pro-260628",
    instruction="Use the `calculator` tool to perform the calculation the user asks for.",
    tools=[calculator],
)
runner = Runner(agent=agent)

response = asyncio.run(runner.run("Add 2 and 3"))
print(response)

Context-aware functions

examples/tools/function_tools/tool_context_usage.py

import asyncio

from google.adk.tools.tool_context import ToolContext
from veadk import Agent, Runner


def message_checker(user_message: str, tool_context: ToolContext) -> str:
    """Check a user message and return it normalized.

    Args:
        user_message (str): The user message to check.

    Returns:
        str: The checked message.
    """
    call_count = tool_context.state.get("message_checker_calls", 0) + 1
    tool_context.state["message_checker_calls"] = call_count

    return f"Checked message: {user_message.upper()} (call {call_count})"


agent = Agent(
    name="context_agent",
    model_name="doubao-seed-2-1-pro-260628",
    instruction="Use message_checker to check the user message, then show the checked result.",
    tools=[message_checker],
)
runner = Runner(agent=agent)

response = asyncio.run(runner.run("Hello world!"))
print(response)

The tool function can read and update shared session state through tool_context.state, so later tool calls and callbacks can reuse that information.

Long-running tasks

This example simulates submission and completion without downloading data or starting a background job. The tool returns pending, the run loop captures the long-running call, then a FunctionResponse feeds back a finish status so the agent can produce its final reply.

examples/tools/function_tools/long_running_tool.py

import asyncio
from typing import Any

from google.adk.events import Event
from google.adk.tools.long_running_tool import LongRunningFunctionTool
from google.genai.types import Content, FunctionCall, FunctionResponse, Part
from veadk import Agent, Runner

APP_NAME = "long_running_tool_app"
USER_ID = "long_running_tool_user"
SESSION_ID = "long_running_tool_session"


def big_data_processing(data_url: str) -> dict[str, Any]:
    """Start processing big data located at a URL.

    Args:
        data_url (str): The URL of the big data to process.

    Returns:
        dict[str, Any]: The initial task state, with "status" == "pending",
        the "data-url", and a "task-id" to track the job.
    """
    # Simulate a submitted job; replace this with a real job service.
    return {
        "status": "pending",
        "data-url": data_url,
        "task-id": "big-data-processing-1",
    }


long_running_tool = LongRunningFunctionTool(func=big_data_processing)

agent = Agent(
    name="long_running_tool_agent",
    model_name="doubao-seed-2-1-pro-260628",
    instruction="Use big_data_processing to process big data.",
    tools=[long_running_tool],
)
runner = Runner(agent=agent, app_name=APP_NAME)


def get_long_running_call(event: Event) -> FunctionCall | None:
    """Return the long-running function call carried by an event, if any."""
    if not event.long_running_tool_ids or not event.content or not event.content.parts:
        return None
    for part in event.content.parts:
        if (
            part.function_call
            and part.function_call.id in event.long_running_tool_ids
        ):
            return part.function_call
    return None


def get_function_response(
    event: Event, function_call_id: str
) -> FunctionResponse | None:
    """Return the function response matching a given call id, if any."""
    if not event.content or not event.content.parts:
        return None
    for part in event.content.parts:
        if (
            part.function_response
            and part.function_response.id == function_call_id
        ):
            return part.function_response
    return None


async def main():
    # Create the session before running.
    session = await runner.short_term_memory.create_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )

    query = "Process the big data from https://example.com/data.csv"
    content = Content(role="user", parts=[Part(text=query)])

    print("Running agent...")
    long_running_call = None
    long_running_response = None
    async for event in runner.run_async(
        session_id=session.id, user_id=USER_ID, new_message=content
    ):
        if long_running_call is None:
            long_running_call = get_long_running_call(event)
        elif long_running_response is None:
            long_running_response = get_function_response(event, long_running_call.id)
        if event.content and event.content.parts:
            if text := "".join(part.text or "" for part in event.content.parts):
                print(f"[{event.author}]: {text}")

    # Simulate completion and send the final result back to the agent.
    if long_running_response is not None:
        updated = long_running_response.model_copy(deep=True)
        updated.response = {"status": "finish"}
        async for event in runner.run_async(
            session_id=session.id,
            user_id=USER_ID,
            new_message=Content(
                role="user", parts=[Part(function_response=updated)]
            ),
        ):
            if event.content and event.content.parts:
                if text := "".join(part.text or "" for part in event.content.parts):
                    print(f"[{event.author}]: {text}")


if __name__ == "__main__":
    asyncio.run(main())

Synchronous, asynchronous, and long-running calls

A normal tool may use def or async def. Async functions can await network or storage operations, but do not automatically create background jobs. LongRunningFunctionTool marks a call that expects a later result; the application still owns submission, persistence, polling, and failure handling. When supplying the final result, preserve the original FunctionResponse call ID and tool name, and use the same application, user, and session identifiers. Do not attach another request’s result. Normal functions should also return an explicit success or failure status so error text is not treated as business data.

Overview

Get started

Core components

Observability

Extensions

Frontend

Security

Deployment

Overview

Prerequisites

Usage

Plain-argument functions

Context-aware functions

Long-running tasks

Synchronous, asynchronous, and long-running calls

​Overview

​Prerequisites

​Usage

​Plain-argument functions

​Context-aware functions

​Long-running tasks

​Synchronous, asynchronous, and long-running calls

Overview

Prerequisites

Usage

Plain-argument functions

Context-aware functions

Long-running tasks

Synchronous, asynchronous, and long-running calls