Building Custom Async Function Tools in Google ADK

Most enterprise AI agents do not operate in isolation. They need to interact with external databases, third-party REST APIs, or internal microservices. When these external network requests take hundreds of milliseconds—or even seconds—blocking the execution thread with synchronous code severely bottlenecks your agent harness.

In this guide, we will explore how to construct asynchronous function tools (async def) in Google ADK to process concurrent I/O operations efficiently without blocking the main event loop.

---

Why Asynchronous Tools Matter

Consider a logistics agent tasked with checking inventory across three distributed warehouse databases.

• Synchronous Execution: A synchronous tool blocks the Python asyncio event loop for the full duration of each SQL query or HTTP request. Multiple independent tool calls execute serially, multiplying latency on every turn.
• Asynchronous Execution: An async tool yields control back to the event loop during I/O wait periods. This allows the application to handle parallel tasks, process concurrent user requests, or stream response chunks without idle CPU delay.

flowchart TB

    subgraph S["Sync Execution (Total: 600ms)"]
        direction LR
        S1["DB Call 1<br/>300ms"]
        S2["DB Call 2<br/>300ms"]
        S1 --> S2
    end

    subgraph A["Async Execution (Total: 300ms)"]
        direction LR

        Start((Start))

        Start --> A1["DB Call 1<br/>300ms"]
        Start --> A2["DB Call 2<br/>300ms"]

        A1 --> End((End))
        A2 --> End
    end

    classDef sync fill:#fee2e2,stroke:#dc2626,stroke-width:1.5px,color:#111827;
    classDef async fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#111827;
    classDef terminal fill:#f3f4f6,stroke:#6b7280,stroke-width:1.5px,color:#111827;

    class S1,S2 sync;
    class A1,A2 async;
    class Start,End terminal;

---

Defining an Async Function Tool in ADK

Writing an asynchronous tool in Google ADK follows the same standard pattern as synchronous functions, with the key difference being the async def signature and await keywords inside the implementation.

import asyncio
import httpx
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

async def fetch_product_inventory(sku: str) -> dict:
    """Asynchronously fetches current warehouse stock levels for a given SKU.

    Args:
        sku: The Stock Keeping Unit identifier (e.g. 'PROD-998').
    """
    # Simulate a non-blocking asynchronous database or HTTP lookup
    await asyncio.sleep(0.5)

    mock_db = {
        "PROD-998": {"name": "Ergonomic Keyboard", "stock": 42, "warehouse": "US-East"},
        "PROD-102": {"name": "Wireless Mouse", "stock": 0, "warehouse": "US-West"},
    }

    if sku in mock_db:
        return {"found": True, "details": mock_db[sku]}
    return {"found": False, "error": f"SKU {sku} not recognized"}

# ADK automatically inspects the async signature
async_inventory_tool = FunctionTool(func=fetch_product_inventory)

Google ADK automatically inspects the function signature. Upon discovering an async def coroutine function, the harness schedules its execution directly onto the running asyncio event loop.

---

Wiring Async Tools into the Agent Harness

Connecting an asynchronous tool to an ADK Agent and running it via Runner is seamless:

agent = Agent(
    name="inventory_agent",
    model="gemini-2.5-flash",
    instruction=(
        "You are an automated logistics assistant. "
        "Use `fetch_product_inventory` to check item stock availability."
    ),
    tools=[async_inventory_tool]
)

async def main():
    session_service = InMemorySessionService()
    runner = Runner(agent=agent, session_service=session_service)

    print("Submitting inventory check request...")
    response = await runner.run_async(
        session_id="inv-session-1",
        message="Do we have any stock left for item PROD-998?"
    )
    
    print("\n--- Agent Response ---")
    print(response.text)

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

---

Production Pattern: Non-Blocking HTTP Clients

In real-world applications, avoid synchronous libraries like requests inside async tools. Instead, utilize httpx.AsyncClient or aiohttp to perform network requests asynchronously:

async def fetch_real_inventory(sku: str) -> dict:
    """Fetches real-time stock levels from an external inventory microservice.

    Args:
        sku: Product SKU identifier.
    """
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            response = await client.get(f"https://api.warehouse.internal/v1/stock/{sku}")
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as err:
            return {"found": False, "error": f"HTTP error occurred: {err.response.status_code}"}
        except httpx.RequestError as err:
            return {"found": False, "error": f"Network error connecting to inventory service: {err}"}

---

Critical Rules for Docstrings & Type Annotations

Whether your function tools are synchronous or asynchronous, Google ADK relies strictly on static analysis of python docstrings and type hints to generate OpenAPI tool schemas for foundation models.

[!IMPORTANT] To prevent Gemini from misinterpreting tool parameters or generating invalid function call arguments, ensure your async tool definitions include:

1. Explicit Function Summary: A clear explanation of what task the tool performs.
2. Type Annotations: Explicit Python type hints for every input argument and return type (sku: str -> dict).
3. Google-Style Argument Descriptions: Detailed Args: block detailing format requirements, constraints, and examples.

---

Key Takeaway

Asynchronous function tools are a direct drop-in replacement for synchronous tools in Google ADK. Whenever your tool performs network I/O, database access, or file system operations, adopt async def and httpx to keep your production agent harness responsive and scalable under high concurrent loads.