LangChain Tutorial (Python): handling async tools for advanced developers

This tutorial shows you how to build a LangChain agent that can call async tools correctly in Python, without blocking the event loop or mixing sync and async execution paths. You need this when your tools hit HTTP APIs, databases, queues, or internal services that already expose async clients and you want the agent to stay responsive under load.

What You'll Need

• •Python 3.10+
• •langchain
• •langchain-openai
• •pydantic
• •python-dotenv
• •An OpenAI API key set as OPENAI_API_KEY
• •A terminal and a clean virtual environment

Install the packages:

pip install langchain langchain-openai pydantic python-dotenv

Step-by-Step

1. •Start with a tool that has an async implementation.
   In LangChain, the cleanest path is to define both sync and async entry points on the same tool so your agent can use it in either mode.

import asyncio
from typing import Optional

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field


class WeatherInput(BaseModel):
    city: str = Field(..., description="City name")
    units: Optional[str] = Field(default="metric", description="metric or imperial")


class AsyncWeatherTool(BaseTool):
    name: str = "weather_lookup"
    description: str = "Get weather for a city"
    args_schema = WeatherInput

    def _run(self, city: str, units: str = "metric") -> str:
        return f"{city}: 21°C, clear sky ({units})"

    async def _arun(self, city: str, units: str = "metric") -> str:
        await asyncio.sleep(0.2)
        return f"{city}: 21°C, clear sky ({units})"

2. •Wire the tool into an agent that supports function calling.
   The important part here is using create_openai_tools_agent, which knows how to invoke tools from model output and works well with async execution when you call the chain asynchronously.

import os
from dotenv import load_dotenv

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [AsyncWeatherTool()]

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a precise assistant. Use tools when needed."),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)

agent = create_openai_tools_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

3. •Call the agent with ainvoke, not invoke.
   If your tool is async and you are already inside an async app or service, this keeps the whole stack non-blocking.

import asyncio


async def main():
    result = await executor.ainvoke(
        {"input": "What is the weather in Nairobi?"},
    )
    print(result["output"])


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

4. •If you need direct tool access outside the agent loop, call _arun explicitly through an async wrapper.
   This is useful for testing tool behavior in isolation before plugging it into an agent.

import asyncio


async def test_tool():
    tool = AsyncWeatherTool()
    result = await tool._arun(city="Lagos", units="metric")
    print(result)


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

5. •When your real tool hits external systems, keep network I/O inside _arun and avoid blocking libraries there.
   If you must use a sync SDK that has no async client, run it in a thread executor so you do not freeze the event loop.

import asyncio


class WrappedSyncTool(AsyncWeatherTool):
    def _run(self, city: str, units: str = "metric") -> str:
        return f"{city}: fetched via sync SDK"

    async def _arun(self, city: str, units: str = "metric") -> str:
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None,
            lambda: self._run(city=city, units=units),
        )

Testing It

Run the script and confirm that verbose=True shows the agent deciding whether to call the tool. If everything is wired correctly, you should see the final answer printed after the tool result comes back through ainvoke.

A good sanity check is to replace _arun with an artificial delay like asyncio.sleep(1) and verify your app still stays responsive if you have other concurrent tasks running. Also confirm that calling invoke() on this setup from inside an existing event loop does not become your default path; use ainvoke() for async flows.

If you are integrating this into FastAPI or another async web framework, run multiple requests concurrently and watch for head-of-line blocking. The agent should complete each request independently as long as your tool code avoids synchronous I/O in _arun.

Next Steps

• •Add structured outputs with Pydantic so your async tools return typed data instead of raw strings.
• •Learn how to batch multiple tool calls with abatch() when one user request fans out across several services.
• •Move from a single tool to a router pattern where different async tools handle CRM, policy lookup, claims status, or payment checks.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.