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

This tutorial shows how to wire async tools into an AutoGen agent workflow in Python without blocking the event loop. You need this when your agent calls network APIs, databases, or long-running services and you want concurrency instead of serial waits.

What You'll Need

• •Python 3.10+
• •autogen-agentchat
• •autogen-ext[openai]
• •An OpenAI API key in OPENAI_API_KEY
• •Basic familiarity with AutoGen agents and tool calling
• •Access to an async-capable tool you want to expose, or the sample HTTP tool below

Step-by-Step

1. •Install the packages and set up your environment.
   The important part here is using the AgentChat stack plus the OpenAI extension package, because that gives you a clean path for tool-enabled agents.

pip install autogen-agentchat autogen-ext[openai] httpx
export OPENAI_API_KEY="your-key-here"

2. •Define an async tool that does real I/O.
   This example uses httpx.AsyncClient so you can see the exact pattern AutoGen needs: an async def function with a typed signature and a return value that can be serialized.

import asyncio
import httpx

async def fetch_status(url: str) -> str:
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(url)
        return f"{url} -> {response.status_code}"

3. •Wrap the async function as an AutoGen tool and attach it to an agent.
   FunctionTool is the key piece here. AutoGen can call your coroutine directly, and the model will see the tool schema generated from your function signature.

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.tools import FunctionTool

from tools import fetch_status  # if you split files; otherwise use the function inline

async_tool = FunctionTool(fetch_status, description="Fetch HTTP status for a URL")

model_client = OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],
)

agent = AssistantAgent(
    name="assistant",
    model_client=model_client,
    tools=[async_tool],
)

4. •Run the agent inside an async entrypoint and let it decide when to call the tool.
   Do not try to call agent.run() from sync code unless you are already inside an event loop strategy you control. For production services, keep everything async end-to-end.

import os
import asyncio
from autogen_agentchat.messages import TextMessage

async def main() -> None:
    result = await agent.run(
        task="Check https://example.com and tell me whether it is reachable."
    )

    for message in result.messages:
        print(type(message).__name__, getattr(message, "content", ""))

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

5. •If you have multiple async tools, keep them independent and let AutoGen schedule them through separate calls.
   The pattern below shows a second tool for reading a URL body length; in real systems this could be a pricing API, policy service, or claims lookup endpoint.

import httpx
from autogen_core.tools import FunctionTool

async def fetch_length(url: str) -> int:
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(url)
        return len(response.text)

length_tool = FunctionTool(fetch_length, description="Return response body length for a URL")

agent = AssistantAgent(
    name="assistant",
    model_client=model_client,
    tools=[async_tool, length_tool],
)

Testing It

Run the script against a stable public endpoint like https://example.com first. You should see the agent choose one of the tools, execute it without blocking, and return a response that includes the HTTP status or body length.

If it hangs, check three things: your event loop usage, your API key, and whether the tool function is truly declared with async def. If AutoGen cannot infer the schema cleanly, make sure every parameter has a type annotation and every tool returns JSON-serializable data like strings, numbers, lists, or dicts.

For more confidence, point the tool at two endpoints with different latency and confirm that your app stays responsive while awaiting responses. In a larger service, add logging around each tool call so you can measure latency per dependency.

Next Steps

• •Add timeout handling and retries around each async tool call.
• •Move from single-tool agents to multi-agent workflows where one agent delegates I/O-heavy work.
• •Add structured outputs so downstream systems can consume tool results without parsing free-form text.

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.