Async Tools¶

This notebook demonstrates AsyncSimpleFunctionTool and AsyncPydanticFunctionTool — the async counterparts to the sync tools introduced in Chapter 2.

Async tools wrap async functions and are invoked with await. They are the right choice whenever the underlying operation is I/O-bound (network calls, database queries, file reads), because they avoid blocking the event loop while waiting for a response.

No LLM is involved — this is pure Chapter 2 tool mechanics.

In [ ]:

# Uncomment the line below to install `llm-agents-from-scratch` from PyPI
# !pip install llm-agents-from-scratch

# Uncomment the line below to install `llm-agents-from-scratch` from PyPI # !pip install llm-agents-from-scratch

AsyncSimpleFunctionTool¶

AsyncSimpleFunctionTool wraps any async function the same way SimpleFunctionTool wraps a sync one. Here we wrap a function that fetches the current weather temperature for a location from the Open-Meteo API — a free, no-auth endpoint.

We use asyncio.to_thread to run the blocking urllib call off the main event loop, which is the standard pattern for wrapping sync I/O in async code without adding extra dependencies.

In [1]:

import asyncio
import json
import urllib.request

from llm_agents_from_scratch.data_structures import ToolCall
from llm_agents_from_scratch.tools import AsyncSimpleFunctionTool


async def get_temperature(latitude: float, longitude: float) -> str:
    """Fetch the current temperature (°C) for a lat/lon coordinate."""
    url = (
        "https://api.open-meteo.com/v1/forecast"
        f"?latitude={latitude}&longitude={longitude}"
        "&current_weather=true"
    )
    req = urllib.request.Request(
        url,
        headers={"User-Agent": "llm-agents-from-scratch/1.0"},
    )
    raw = await asyncio.to_thread(
        lambda: urllib.request.urlopen(req).read().decode(),
    )
    data = json.loads(raw)
    temp = data["current_weather"]["temperature"]
    return f"{temp} °C"


temperature_tool = AsyncSimpleFunctionTool(func=get_temperature)
print(f"Tool name:  {temperature_tool.name}")
print(f"Tool desc:  {temperature_tool.description}")

import asyncio import json import urllib.request from llm_agents_from_scratch.data_structures import ToolCall from llm_agents_from_scratch.tools import AsyncSimpleFunctionTool async def get_temperature(latitude: float, longitude: float) -> str: """Fetch the current temperature (°C) for a lat/lon coordinate.""" url = ( "https://api.open-meteo.com/v1/forecast" f"?latitude={latitude}&longitude={longitude}" "&current_weather=true" ) req = urllib.request.Request( url, headers={"User-Agent": "llm-agents-from-scratch/1.0"}, ) raw = await asyncio.to_thread( lambda: urllib.request.urlopen(req).read().decode(), ) data = json.loads(raw) temp = data["current_weather"]["temperature"] return f"{temp} °C" temperature_tool = AsyncSimpleFunctionTool(func=get_temperature) print(f"Tool name: {temperature_tool.name}") print(f"Tool desc: {temperature_tool.description}")

Tool name:  get_temperature
Tool desc:  Fetch the current temperature (°C) for a lat/lon coordinate.

In [2]:

# London: 51.5°N, -0.1°E
tool_call = ToolCall(
    tool_name="get_temperature",
    arguments={"latitude": 51.5, "longitude": -0.1},
)

result = await temperature_tool(tool_call)
print(f"error:    {result.error}")
print(f"content:  {result.content}")

# London: 51.5°N, -0.1°E tool_call = ToolCall( tool_name="get_temperature", arguments={"latitude": 51.5, "longitude": -0.1}, ) result = await temperature_tool(tool_call) print(f"error: {result.error}") print(f"content: {result.content}")

error:    False
content:  10.5 °C

AsyncPydanticFunctionTool¶

AsyncPydanticFunctionTool is the async version of PydanticFunctionTool: the wrapped function must be async and must accept a single pydantic.BaseModel argument. Pydantic validates the incoming arguments before the function is called.

Here we fetch sunrise and sunset times for a location from the Sunrise-Sunset API — also free and no-auth.

In [3]:

from pydantic import BaseModel, Field

from llm_agents_from_scratch.tools import AsyncPydanticFunctionTool


class SunParams(BaseModel):
    """Parameters for the sunrise/sunset lookup tool."""

    latitude: float = Field(description="Latitude of the location.")
    longitude: float = Field(description="Longitude of the location.")
    date: str = Field(
        description="Date in YYYY-MM-DD format. Use 'today' for today.",
    )


async def get_sun_times(params: SunParams) -> str:
    """Return sunrise and sunset times (UTC) for a given location and date."""
    url = (
        "https://api.sunrise-sunset.org/json"
        f"?lat={params.latitude}&lng={params.longitude}"
        f"&date={params.date}&formatted=0"
    )
    req = urllib.request.Request(
        url,
        headers={"User-Agent": "llm-agents-from-scratch/1.0"},
    )
    raw = await asyncio.to_thread(
        lambda: urllib.request.urlopen(req).read().decode(),
    )
    data = json.loads(raw)["results"]
    return json.dumps(
        {"sunrise": data["sunrise"], "sunset": data["sunset"]},
        indent=2,
    )


sun_tool = AsyncPydanticFunctionTool(func=get_sun_times)
print(f"Tool name:  {sun_tool.name}")
print(f"Tool desc:  {sun_tool.description}")

from pydantic import BaseModel, Field from llm_agents_from_scratch.tools import AsyncPydanticFunctionTool class SunParams(BaseModel): """Parameters for the sunrise/sunset lookup tool.""" latitude: float = Field(description="Latitude of the location.") longitude: float = Field(description="Longitude of the location.") date: str = Field( description="Date in YYYY-MM-DD format. Use 'today' for today.", ) async def get_sun_times(params: SunParams) -> str: """Return sunrise and sunset times (UTC) for a given location and date.""" url = ( "https://api.sunrise-sunset.org/json" f"?lat={params.latitude}&lng={params.longitude}" f"&date={params.date}&formatted=0" ) req = urllib.request.Request( url, headers={"User-Agent": "llm-agents-from-scratch/1.0"}, ) raw = await asyncio.to_thread( lambda: urllib.request.urlopen(req).read().decode(), ) data = json.loads(raw)["results"] return json.dumps( {"sunrise": data["sunrise"], "sunset": data["sunset"]}, indent=2, ) sun_tool = AsyncPydanticFunctionTool(func=get_sun_times) print(f"Tool name: {sun_tool.name}") print(f"Tool desc: {sun_tool.description}")

Tool name:  get_sun_times
Tool desc:  Return sunrise and sunset times (UTC) for a given location and date.

In [4]:

# Tokyo: 35.7°N, 139.7°E
tool_call = ToolCall(
    tool_name="get_sun_times",
    arguments={"latitude": 35.7, "longitude": 139.7, "date": "today"},
)

result = await sun_tool(tool_call)
print(f"error:    {result.error}")
print(f"content:\n{result.content}")

# Tokyo: 35.7°N, 139.7°E tool_call = ToolCall( tool_name="get_sun_times", arguments={"latitude": 35.7, "longitude": 139.7, "date": "today"}, ) result = await sun_tool(tool_call) print(f"error: {result.error}") print(f"content:\n{result.content}")

error:    False
content:
{
  "sunrise": "2026-05-07T19:40:56+00:00",
  "sunset": "2026-05-08T09:34:30+00:00"
}

Sync vs Async — Call Pattern Comparison¶

The only difference at the call site is await. The ToolCall object and ToolCallResult shape are identical, which is what will allow LLMAgent to handle both tool types transparently.

In [5]:

from llm_agents_from_scratch.tools import SimpleFunctionTool


def hailstone_step_func(x: int) -> int:
    """Perform a single step of the Hailstone (Collatz) sequence."""
    if x % 2 == 0:
        return x // 2
    return 3 * x + 1


async def async_hailstone_step_func(x: int) -> int:
    """Perform a single step of the Hailstone (Collatz) sequence."""
    return await asyncio.to_thread(hailstone_step_func, x)


sync_tool = SimpleFunctionTool(func=hailstone_step_func)
async_tool = AsyncSimpleFunctionTool(func=async_hailstone_step_func)

tool_call = ToolCall(tool_name="hailstone_step_func", arguments={"x": 12})

sync_result = sync_tool(tool_call)  # sync — no await
async_result = await async_tool(tool_call)  # async — requires await

print(f"sync  result: {sync_result.content}  (error={sync_result.error})")
print(f"async result: {async_result.content}  (error={async_result.error})")

from llm_agents_from_scratch.tools import SimpleFunctionTool def hailstone_step_func(x: int) -> int: """Perform a single step of the Hailstone (Collatz) sequence.""" if x % 2 == 0: return x // 2 return 3 * x + 1 async def async_hailstone_step_func(x: int) -> int: """Perform a single step of the Hailstone (Collatz) sequence.""" return await asyncio.to_thread(hailstone_step_func, x) sync_tool = SimpleFunctionTool(func=hailstone_step_func) async_tool = AsyncSimpleFunctionTool(func=async_hailstone_step_func) tool_call = ToolCall(tool_name="hailstone_step_func", arguments={"x": 12}) sync_result = sync_tool(tool_call) # sync — no await async_result = await async_tool(tool_call) # async — requires await print(f"sync result: {sync_result.content} (error={sync_result.error})") print(f"async result: {async_result.content} (error={async_result.error})")

sync  result: 6  (error=False)
async result: 6  (error=False)