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.

Setting up the LLM¶

I recommend running these notebooks on Ollama Cloud. It serves much larger models than most laptops can run. The default cloud model is picked by benchmarking candidates on these notebooks, and an automated run checks every notebook against it each week. The notebook status page shows the current model and results.

1. Create an API key at https://ollama.com/settings/keys.
2. Export it before launching Jupyter: export OLLAMA_API_KEY=<your key>.
3. That's all. When a notebook builds its LLM, it detects the key and uses Ollama Cloud.

Prefer to run locally? Install Ollama from https://ollama.com/download and pull the model with ollama pull qwen3:14b. If no key is set, the notebook starts the Ollama service for you, so there is no need to run ollama serve. Local models are smaller, so your outputs will differ from the recorded ones.

Prefer OpenAI or Anthropic?

1. Install the extra: uv sync --extra openai or uv sync --extra anthropic.
2. Export OPENAI_API_KEY or ANTHROPIC_API_KEY before launching Jupyter. If you forget, you will be prompted for it, which keeps the key out of the saved notebook.
3. Set LLM_PROVIDER=openai (or anthropic), or pass provider="openai" to make_llm(). A key on its own never switches providers, so one exported for unrelated work cannot reroute you.

The switch applies wherever a notebook builds its LLM with make_llm(). A notebook that constructs OllamaLLM directly stays on Ollama. Output on gpt-5 or Claude will differ from the recorded outputs, and prompt-sensitive examples (the ch09 evaluator pattern, ch08 supervised trajectories) may behave noticeably differently.

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)