Sending sequential LLM requests wastes latency budget when you need outputs from multiple models. With the openai python sdk async concurrent calls pattern, you can fire GPT-4o and Claude requests in parallel using asyncio and cut wall-clock time to the slowest response. This guide builds a runnable script that targets any OpenAI-compatible endpoint and verifies the speedup.
Step 1: Install the Async-Capable SDK and Dependencies
The official openai package ships an AsyncOpenAI client. You do not need a separate library for concurrency; Python’s asyncio is in the standard library.
pip install "openai>=1.30.0" python-dotenvPin a recent version. The async client has been stable since the 1.x line, but older releases had rougher edges on timeout handling and header forwarding. If you are on Python 3.8 or earlier, upgrade—async support assumptions here target 3.10+ syntax.
Step 2: Set Up Environment and Endpoint Configuration
Keep credentials out of source. Create a .env file with your gateway URL and key.
# .env
OPENAI_API_KEY=sk-your-key
OPENAI_BASE_URL=https://api.openai.com/v1 # or your compatible gateway
MODEL_GPT=gpt-4o
MODEL_CLAUDE=claude-3-5-sonnet-20240620If you run against a gateway that aggregates providers, the model slug may be prefixed (e.g., anthropic/claude-3-5-sonnet). Check your provider’s model list. The OpenAI-compatible contract only requires the model field to be a string; semantics are up to the server.
Load it in Python:
from dotenv import load_dotenv
import os
load_dotenv()
API_KEY = os.environ["OPENAI_API_KEY"]
BASE_URL = os.environ["OPENAI_BASE_URL"]
MODEL_GPT = os.environ["MODEL_GPT"]
MODEL_CLAUDE = os.environ["MODEL_CLAUDE"]Step 3: Initialize the Async Client
The AsyncOpenAI constructor mirrors the sync one. Set base_url to your compatible endpoint and pass the key.
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key=API_KEY,
base_url=BASE_URL,
timeout=30.0, # seconds, per request
max_retries=0, # we handle retries ourselves
)Setting max_retries=0 is opinionated but deliberate: when you run openai python sdk async concurrent calls, you want failures to surface immediately so asyncio.gather can capture them instead of blocking the event loop with hidden backoff. The client is cheap to construct; create one at module scope and reuse it.
Step 4: Define a Single Model Call Coroutine
Wrap one chat completion in an async def. Use response.choices[0].message.content for the text.
async def call_model(client, model, prompt, system="You are a concise assistant."):
resp = await client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
temperature=0.2,
max_tokens=500,
)
return {
"model": model,
"text": resp.choices[0].message.content,
"usage": resp.usage.model_dump() if resp.usage else None,
}Note the usage field. Per-token metering is exposed by compliant gateways; capture it for cost tracking. If you need provider cache hints, pass extra_headers={"cache-control": "max-age=300"} to the create call—compatible gateways forward those to the upstream.
Step 5: Run GPT-4o and Claude Calls Concurrently
The core of the openai python sdk async concurrent calls approach is asyncio.gather. It schedules both coroutines on the running event loop and awaits all of them. The loop yields control while each await is pending, so the HTTP requests are in flight simultaneously.
import asyncio
async def main():
prompt = "Summarize the trade-offs of async versus threaded concurrency in 3 bullet points."
tasks = [
call_model(client, MODEL_GPT, prompt),
call_model(client, MODEL_CLAUDE, prompt),
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
if __name__ == "__main__":
out = asyncio.run(main())
for r in out:
if isinstance(r, Exception):
print("ERROR:", repr(r))
else:
print(f"--- {r['model']} ---")
print(r["text"])
print("usage:", r["usage"])return_exceptions=True prevents one failed model from cancelling the other. You still get both outputs (or an error object) without losing the sibling response. Without it, the first raised exception would propagate and cancel pending tasks.
Step 6: Add Timeouts and Error Isolation
Network calls fail. Wrap each task with asyncio.wait_for to enforce a hard bound independent of the client timeout.
async def call_with_timeout(client, model, prompt, seconds=20):
try:
return await asyncio.wait_for(
call_model(client, model, prompt), timeout=seconds
)
except asyncio.TimeoutError:
return {"model": model, "error": f"timeout after {seconds}s"}
except Exception as e:
return {"model": model, "error": repr(e)}Then in main:
tasks = [
call_with_timeout(client, MODEL_GPT, prompt),
call_with_timeout(client, MODEL_CLAUDE, prompt),
]
results = await asyncio.gather(*tasks)Now each result is a dict with either text or error. No exception escapes gather. This structure is what makes the openai python sdk async concurrent calls pattern safe to drop into a service.
Step 7: Execute and Verify Success
Run the script:
python parallel_calls.pyVerifying Concurrent Execution
To prove the calls ran in parallel, time the script and compare against sequential execution.
import time
async def timed_main():
start = time.monotonic()
results = await main()
elapsed = time.monotonic() - start
print(f"Total wall time: {elapsed:.2f}s")
return results
# in __main__:
out = asyncio.run(timed_main())If both models take ~2s individually, concurrent execution should report ~2–3s total, not ~4–6s. That delta is your verification. Also assert that neither result contains an error key:
assert all("text" in r for r in out), "Some calls failed"Check the printed usage blocks to confirm tokens were consumed from both models. If you see two distinct model names and two non-empty texts, the pipeline works.
Step 8: Use a Gateway with Built-in Fallback
When you point the SDK at n4n.ai, one OpenAI-compatible endpoint addresses 240+ models and automatically falls back when a provider is rate-limited or degraded. That removes manual retry scaffolding for most production cases. The same AsyncOpenAI client works unchanged; just set BASE_URL and API_KEY to the gateway credentials.
The gateway also forwards provider cache-control hints if you pass them in extra_headers, and honors client routing directives via model prefixes. Your concurrent calls stay simple while the gateway handles cross-provider reliability.
Production Considerations
Concurrency limits. Most gateways throttle simultaneous requests per key. Wrap gather in an asyncio.Semaphore if you scale beyond a handful of models.
sem = asyncio.Semaphore(5)
async def call_limited(model, prompt):
async with sem:
return await call_with_timeout(client, model, prompt)Token metering. Capture usage from each response. If your endpoint provides per-token metering, aggregate prompt_tokens and completion_tokens to attribute cost across models. Store these in your logging pipeline.
Cache control. OpenAI-compatible headers like cache-control: max-age=300 can be sent via extra_headers={"cache-control": "max-age=300"} on the create call. Gateways that forward these hints reduce repeat cost on identical prefixes—useful when you send the same system prompt to GPT-4o and Claude.
Model routing. Some gateways let you pin a provider with a prefix (anthropic/, openai/). Use explicit slugs in env vars so the openai python sdk async concurrent calls pattern does not accidentally drift to a different backend during a config change.
Structured output. If you need JSON, set response_format={"type": "json_object"} in the create call. Both models support it; parse with json.loads after the gather.
Final Checklist
-
openaiandpython-dotenvinstalled -
.envholds base URL, key, and model slugs -
AsyncOpenAIclient initialized with explicit timeout - Coroutine wraps a single completion
-
asyncio.gatherwithreturn_exceptionsor per-task error dict - Script timed and shows parallel speedup
- Gateway fallback (if used) confirmed via status logs
That is the complete path from zero to parallel GPT-4o and Claude completions with the OpenAI Python SDK. Adjust model names to your endpoint’s catalog and ship it behind a semaphore when traffic grows.