AutoGen Tutorial (Python): handling async tools for beginners

This tutorial shows you how to wire async Python tools into an AutoGen agent without blocking the event loop. You need this when your agent calls APIs, databases, or internal services that already use async def, and you want the agent to handle them cleanly instead of wrapping everything in brittle sync code.

What You'll Need

• •Python 3.10+
• •autogen-agentchat
• •autogen-ext
• •An OpenAI API key
• •Basic familiarity with AutoGen agents and tool registration
• •A terminal with pip available

Install the packages:

pip install autogen-agentchat autogen-ext openai

Set your API key:

export OPENAI_API_KEY="your-key-here"

Step-by-Step

1. •Start with a simple async function you want the agent to call.
   This example simulates an external service call using asyncio.sleep, which is the same shape as a real HTTP request or database lookup.

import asyncio

async def get_policy_status(policy_id: str) -> str:
    await asyncio.sleep(1)
    return f"Policy {policy_id} is active and paid through 2026-01-31."

2. •Wrap that async function as an AutoGen tool.
   AutoGen tools can be declared from normal Python callables, and async functions are supported directly. The important part is that the tool metadata is explicit so the model knows how to call it.

import asyncio
from autogen_core.tools import FunctionTool

async def get_policy_status(policy_id: str) -> str:
    await asyncio.sleep(1)
    return f"Policy {policy_id} is active and paid through 2026-01-31."

policy_status_tool = FunctionTool(
    get_policy_status,
    description="Get the current status of an insurance policy by policy ID.",
)

3. •Create an assistant agent and give it the tool.
   Here we use AssistantAgent from autogen_agentchat. The model can now decide when to call your async tool instead of guessing an answer.

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

async def get_policy_status(policy_id: str) -> str:
    await asyncio.sleep(1)
    return f"Policy {policy_id} is active and paid through 2026-01-31."

policy_status_tool = FunctionTool(
    get_policy_status,
    description="Get the current status of an insurance policy by policy ID.",
)

model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")

agent = AssistantAgent(
    name="insurance_assistant",
    model_client=model_client,
    tools=[policy_status_tool],
)

4. •Run the agent inside an async entrypoint.
   This is where beginners usually get stuck: if your tool is async, your script should also run asynchronously. Use asyncio.run() and await the agent response.

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

async def get_policy_status(policy_id: str) -> str:
    await asyncio.sleep(1)
    return f"Policy {policy_id} is active and paid through 2026-01-31."

policy_status_tool = FunctionTool(
    get_policy_status,
    description="Get the current status of an insurance policy by policy ID.",
)

model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")

agent = AssistantAgent(
    name="insurance_assistant",
    model_client=model_client,
    tools=[policy_status_tool],
)

async def main():
    result = await agent.run(
        task=TextMessage(content="Check policy ABC123 and tell me if it's active.", source="user")
    )
    print(result.messages[-1].content)

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

5. •Add one more async tool so you can see multiple tools work the same way.
   In production, this pattern maps cleanly to separate service calls like claims lookup, customer profile fetch, or payment verification.

import asyncio
from autogen_core.tools import FunctionTool

async def get_claim_count(customer_id: str) -> str:
    await asyncio.sleep(1)
    return f"Customer {customer_id} has 2 open claims."

claim_count_tool = FunctionTool(
    get_claim_count,
    description="Get the number of open claims for a customer.",
)

You can register both tools on the same agent:

agent = AssistantAgent(
    name="insurance_assistant",
    model_client=model_client,
    tools=[policy_status_tool, claim_count_tool],
)

Testing It

Run the script and ask a question that clearly maps to one of your tools, such as “Check policy ABC123 and tell me if it’s active.” If everything is wired correctly, the agent should call the async function and respond with the returned text.

If you want to confirm it’s really asynchronous, keep the await asyncio.sleep(1) delay in place and run two requests in separate tasks later. The point is not speed here; it’s proving your tool doesn’t block other async work.

If you see a model error, check three things first: your OPENAI_API_KEY, your installed package versions, and whether you used asyncio.run() around the entrypoint. Most beginner failures come from mixing sync execution with async tools.

Next Steps

• •Learn how to pass structured inputs with Pydantic models instead of plain strings.
• •Add retries and timeouts around real API calls before exposing them as tools.
• •Move from a single assistant to a multi-agent workflow when tool usage gets more complex.

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.