CrewAI Tutorial (Python): handling async tools for intermediate developers

This tutorial shows you how to wire async Python tools into a CrewAI workflow without blocking your agent loop. You need this when your tool calls hit APIs, databases, or internal services that already expose async def methods and you want CrewAI to call them cleanly from sync agent execution.

What You'll Need

• •Python 3.10+
• •crewai
• •crewai-tools
• •python-dotenv
• •An OpenAI API key in OPENAI_API_KEY
• •Basic familiarity with:
  • •Agent
  • •Task
  • •Crew
  • •custom tools in CrewAI

Install the packages:

pip install crewai crewai-tools python-dotenv

Step-by-Step

1. •Start by creating a tool class that exposes an async implementation. CrewAI tools are usually called from synchronous agent code, so the important part is making sure your async method is wrapped correctly.

import asyncio
from crewai.tools import BaseTool
from pydantic import BaseModel, Field


class WeatherInput(BaseModel):
    city: str = Field(..., description="City name to fetch weather for")


class AsyncWeatherTool(BaseTool):
    name: str = "async_weather_tool"
    description: str = "Fetches weather data asynchronously for a given city"
    args_schema = WeatherInput

    async def _arun(self, city: str) -> str:
        await asyncio.sleep(1)
        return f"Weather for {city}: 22°C and sunny"

    def _run(self, city: str) -> str:
        return asyncio.run(self._arun(city))

2. •Next, define an agent and attach the tool. The key detail here is that the agent does not care whether your tool is backed by async logic as long as the tool implements _run and/or _arun correctly.

import os
from dotenv import load_dotenv
from crewai import Agent

load_dotenv()

weather_tool = AsyncWeatherTool()

research_agent = Agent(
    role="Research Analyst",
    goal="Get accurate weather information quickly",
    backstory="You collect external data using tools and summarize it clearly.",
    tools=[weather_tool],
    verbose=True,
)

3. •Then create a task that forces the agent to use the tool. Keep the prompt specific so you can see whether the tool output is being used instead of hallucinated content.

from crewai import Task

weather_task = Task(
    description=(
        "Use the async weather tool to get the current weather for Nairobi. "
        "Return only the weather summary."
    ),
    expected_output="A short weather summary for Nairobi",
    agent=research_agent,
)

4. •Now assemble the crew and run it. This is where you verify that CrewAI can invoke your async-backed tool through normal agent execution.

from crewai import Crew, Process

crew = Crew(
    agents=[research_agent],
    tasks=[weather_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)

5. •If your real async integration uses HTTP calls, database drivers, or internal SDKs, keep that logic inside _arun. For libraries that are already async-native, do not rewrite them into sync code unless you have to; just bridge them at the tool boundary.

import aiohttp


class AsyncHttpTool(BaseTool):
    name: str = "async_http_tool"
    description: str = "Fetches JSON from an HTTP endpoint asynchronously"
    args_schema = WeatherInput

    async def _arun(self, city: str) -> str:
        url = f"https://example.com/weather?city={city}"
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                data = await response.json()
                return data.get("summary", "No summary returned")

    def _run(self, city: str) -> str:
        return asyncio.run(self._arun(city))

Testing It

Run the script and watch for two things: first, the agent should print verbose logs showing it selected the tool; second, the final output should contain the value returned by _arun, not a made-up answer. If you see event-loop errors like “asyncio.run() cannot be called from a running event loop,” that means you’re trying to nest sync wrappers inside an already-running loop.

For local verification, start with the fake asyncio.sleep() version before wiring real network calls. Once that works, swap in your actual API client and confirm latency drops compared to blocking calls in your own service logs.

Next Steps

• •Add retries and timeouts inside _arun for flaky external systems.
• •Wrap multiple async tools behind one shared client session or connection pool.
• •Move from a single sequential crew to multi-agent workflows once your tool layer is stable.

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.