Building responsive LLM apps means never blocking the event loop. This guide covers async function calling python openai-python patterns that survive real traffic: defining tools, running concurrent tool executions, and looping until the model finishes. We assume Python 3.10+ and openai-python 1.x.
Step 1: Install and import dependencies
Install the current openai package. asyncio ships with the standard library, so no extra install is needed for concurrency.
pip install "openai>=1.30.0"Import what you need. Keep the surface area small:
import asyncio
import json
from openai import AsyncOpenAI
# If you later swap to a gateway, only this line changes.
client = AsyncOpenAI() # or AsyncOpenAI(base_url="https://api.n4n.ai/v1")The AsyncOpenAI client mirrors the sync client but every network call returns a coroutine. Treat it as immutable after creation; it manages its own connection pool.
Step 2: Define tool schemas and async executors
Function calling works by giving the model a JSON Schema describing each function. You then write the actual Python implementation as an async function. Decouple the schema from the executor so you can register multiple tools in a dict.
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current temperature for a city",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string"}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "Get latest stock price for a ticker",
"parameters": {
"type": "object",
"properties": {
"ticker": {"type": "string"}
},
"required": ["ticker"]
}
}
}
]
async def get_weather(city: str) -> str:
# Fake IO-bound call; replace with aiohttp/ httpx.AsyncClient
await asyncio.sleep(0.1)
return f"72F in {city}"
async def get_stock_price(ticker: str) -> str:
await asyncio.sleep(0.1)
return f"{ticker} at $123.45"
executors = {
"get_weather": get_weather,
"get_stock_price": get_stock_price
}The schema is what the model sees. The executor dict is what your code uses to dispatch. Never trust the model: validate arguments before calling.
Step 3: Initialize the client and set the model
If you point the client at an OpenAI-compatible endpoint such as n4n.ai, the same async function calling python openai-python code gains automatic fallback when a provider is degraded, without extra logic. For local testing, the default client works against OpenAI’s API.
MODEL = "gpt-4o-mini" # or any model your endpoint supportsKeep the model name in one constant. In production you will want to inject it via env var so you can shift models without code changes.
Step 4: Build the conversation loop
The core pattern is a loop: send messages, check for tool calls, execute them, append results, repeat. The loop ends when the model returns a message with no tool calls.
async def run_conversation(user_prompt: str) -> str:
messages = [{"role": "user", "content": user_prompt}]
while True:
resp = await client.chat.completions.create(
model=MODEL,
messages=messages,
tools=tools,
tool_choice="auto"
)
msg = resp.choices[0].message
messages.append(msg) # append the assistant message (may contain tool_calls)
if not msg.tool_calls:
return msg.content or ""
# Step 5 handles the branch below
results = await dispatch_tools(msg.tool_calls)
messages.extend(results)Appending msg directly works because openai-python serializes the tool_calls field correctly when you send it back. Do not try to manually reconstruct the assistant message.
Step 5: Dispatch tool calls concurrently
A single model turn can request multiple tools. Blocking on them sequentially wastes latency. Gather them with asyncio.gather.
async def dispatch_tools(tool_calls) -> list[dict]:
async def handle(tc):
name = tc.function.name
args = json.loads(tc.function.arguments)
if name not in executors:
content = f"error: unknown tool {name}"
else:
try:
content = await executors[name](**args)
except Exception as e:
content = f"error: {e}"
return {
"role": "tool",
"tool_call_id": tc.id,
"content": content
}
return await asyncio.gather(*(handle(tc) for tc in tool_calls))This is the core advantage of async function calling python openai-python over a sync script: if the model asks for weather and a stock price at once, both IO operations run in parallel. Each returned dict carries the tool_call_id so the model can match results to calls.
Step 6: Feed results back and complete
The loop in Step 4 already appends the tool results and calls the model again. On the second pass, the model should produce a final natural-language answer. If it instead emits more tool calls, the loop continues. Cap iterations to avoid infinite loops from a misbehaving model:
MAX_TURNS = 5
async def run_conversation(user_prompt: str) -> str:
messages = [{"role": "user", "content": user_prompt}]
for _ in range(MAX_TURNS):
resp = await client.chat.completions.create(
model=MODEL,
messages=messages,
tools=tools,
tool_choice="auto"
)
msg = resp.choices[0].message
messages.append(msg)
if not msg.tool_calls:
return msg.content or ""
results = await dispatch_tools(msg.tool_calls)
messages.extend(results)
raise RuntimeError("Exceeded MAX_TURNS without final answer")Step 7: Run and verify success
Wire a main and execute with asyncio.run. Verification is straightforward: print the final string and assert that no tool calls remain in the last message.
async def main():
answer = await run_conversation(
"What's the weather in Tokyo and the price of AAPL?"
)
print("FINAL:", answer)
assert isinstance(answer, str) and len(answer) > 0
# Optional: inspect token usage
# print(resp.usage) # if you capture it
if __name__ == "__main__":
asyncio.run(main())Run the script. Success looks like a single printed FINAL line that combines both tool outputs into a coherent sentence. If you see a RuntimeError about MAX_TURNS, the model is looping—tighten your tool descriptions or set tool_choice to force a specific function.
To verify the async path is actually concurrent, add print timestamps inside get_weather and get_stock_price. Both should start within milliseconds of each other, not sequentially.
Production notes
- Timeouts: Wrap
client.chat.completions.createwithasyncio.wait_foror settimeouton the client. A hung provider should not stall your event loop. - Retries: openai-python has a retry config, but for async you may want custom backoff on
APIConnectionError. - Concurrency limits: If you fan out to dozens of tools, bound
asyncio.gatherwith aasyncio.Semaphoreto avoid overwhelming downstream APIs. - Streaming: For long final answers, use
stream=Trueand iterateasync for chunk. Tool calls still arrive in the first chunk set; buffer them before reacting. - Schema drift: Pin your JSON Schema and validate with
pydanticbefore calling executors. The model will occasionally send malformed args.
The async function calling python openai-python loop above is the same structure we use for high-throughput gateways: a single coroutine per request, concurrent tool IO, and a hard cap on turns. Swap the base_url and your code runs against any compatible backend without rewrites.