A missing client-side deadline is the fastest way to stall an entire asyncio event loop when a model endpoint goes quiet. Using python asyncio wait_for timeout llm api calls lets you bound latency per request and shed load instead of blocking forever, but the wrapper has sharp edges around cancellation and streaming that bite in production.
Step 1: Build an async HTTP client with baseline timeouts
Do not use requests in an asyncio service. Use httpx or aiohttp. httpx exposes an AsyncClient that speaks OpenAI-compatible JSON and lets you set granular network timeouts independent of wait_for.
import httpx
BASE_URL = "https://api.openai.com/v1" # swap for any OpenAI-compatible endpoint
API_KEY = "sk-your-key"
def make_client() -> httpx.AsyncClient:
return httpx.AsyncClient(
base_url=BASE_URL,
headers={"Authorization": f"Bearer {API_KEY}"},
timeout=httpx.Timeout(connect=5.0, read=30.0, write=5.0, pool=5.0),
)The httpx.Timeout above kills a stuck socket read at 30 seconds. That is a network-level safety net. It does not, however, cancel your coroutine tree or enforce a business-level latency budget such as “no user waits more than 2 seconds for the first token.” That is what python asyncio wait_for timeout llm api patterns add.
If you front requests with n4n.ai, its automatic fallback switches providers on degradation, but a slow-but-alive upstream can still exceed your latency budget; wait_for remains the client-side backstop.
Step 2: Wrap the completion call in asyncio.wait_for
Write a small function that posts to /chat/completions and wrap the await in asyncio.wait_for. The timeout argument is a float in seconds. When it elapses, wait_for cancels the inner coroutine and raises asyncio.TimeoutError.
import asyncio
async def complete(prompt: str, client: httpx.AsyncClient, timeout: float = 10.0) -> str:
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 200,
}
try:
resp = await asyncio.wait_for(
client.post("/chat/completions", json=payload),
timeout=timeout,
)
except asyncio.TimeoutError:
# Inner post coroutine is already cancelled by wait_for.
raise
resp.raise_for_status()
data = resp.json()
return data["choices"][0]["message"]["content"]Key point: wait_for cancels the task executing client.post. httpx handles cancellation by closing the connection. You do not need to call task.cancel() yourself.
Step 3: Handle cancellation and avoid leaked tasks
A common bug is catching TimeoutError, then later awaiting the same coroutine again. Once cancelled, a coroutine cannot be reused. Build a fresh call per attempt.
import logging
logger = logging.getLogger(__name__)
async def safe_complete(prompt: str, client: httpx.AsyncClient, budget: float = 8.0) -> str | None:
try:
return await asyncio.wait_for(complete(prompt, client), timeout=budget)
except asyncio.TimeoutError:
logger.warning("LLM call exceeded %.1fs budget", budget)
return NoneIf you need retry, re-enter safe_complete or call complete anew. Never wrap an already-awaited coroutine.
When wait_for times out, it calls task.cancel() on the inner task and waits for it to finish cancelling. In Python 3.8+, the cancellation propagates; in older versions you had to do this manually. Assume modern Python.
Step 4: Apply timeouts to streaming responses
Streaming LLM output complicates python asyncio wait_for timeout llm api usage because the response is not a single await but a long-lived generator. You have two options: enforce a total deadline on the whole stream, or enforce a per-chunk deadline.
Total deadline is simpler and usually correct:
async def stream_complete(prompt: str, client: httpx.AsyncClient, total_timeout: float = 20.0):
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"stream": True,
}
async def _consume():
async with client.stream("POST", "/chat/completions", json=payload) as resp:
resp.raise_for_status()
async for line in resp.aiter_lines():
if line.strip():
yield line
# wrap the consumer coroutine, not the stream context alone
async for line in asyncio.wait_for(_consume(), timeout=total_timeout):
print(line)If you need to abort when a single chunk stalls for more than 2 seconds, wrap the anext call inside the loop with wait_for. That requires manually driving the async generator.
async def stream_with_chunk_deadline(gen, chunk_timeout: float = 2.0):
while True:
try:
line = await asyncio.wait_for(gen.__anext__(), timeout=chunk_timeout)
except StopAsyncIteration:
break
except asyncio.TimeoutError:
await gen.aclose()
raise
yield linePick one model. Mixing both total and per-chunk timeouts creates confusing cascade failures.
Step 5: Compose timeouts with concurrency limits
Unbounded concurrency will get you rate-limited. Put a Semaphore around the wait_for-wrapped call so the event loop stays responsive under load.
sem = asyncio.Semaphore(10)
async def bounded_complete(prompt: str, client: httpx.AsyncClient) -> str | None:
async with sem:
return await safe_complete(prompt, client, budget=8.0)
async def batch(prompts: list[str], client: httpx.AsyncClient) -> list[str | None]:
return await asyncio.gather(*(bounded_complete(p, client) for p in prompts))Each item still respects its own wait_for deadline. If the gateway behind the endpoint (such as n4n.ai) honors client routing directives and forwards provider cache-control hints, a timeout may simply mean the cache missed and the provider was slow; your semaphore prevents that slowness from cascading.
Step 6: Verify the timeout with a fault-injected test
A timeout path that has never been exercised is a lie. Use httpx.MockTransport to simulate a slow upstream and assert that asyncio.TimeoutError (or your wrapped None) surfaces.
import pytest
import httpx
async def slow_handler(request: httpx.Request) -> httpx.Response:
await asyncio.sleep(5)
return httpx.Response(200, json={"choices": [{"message": {"content": "late"}}]})
@pytest.mark.asyncio
async def test_llm_timeout():
client = httpx.AsyncClient(transport=httpx.MockTransport(slow_handler))
with pytest.raises(asyncio.TimeoutError):
await complete("hello", client, timeout=0.2)
# ensure no pending tasks leaked
pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
assert not pendingRun pytest -q and confirm the test passes. If you see pending tasks after the test, your cancellation cleanup is incomplete—usually because you caught CancelledError and swallowed it somewhere in the stack.
Verifying success in production
In a live service, export a counter for safe_complete returning None and alert when the rate crosses 1% of requests. Add a trace span around the wait_for call so you can see whether timeouts cluster on a specific model or route. The code above gives you the hook; wire it to your telemetry.
Step 7: Choose timeout values from p99 latency, not guesses
Pull the real p99 completion latency for your model and set wait_for to p99 + 30%. If p99 is 600 ms, a 2-second budget is reasonable. Setting it to 10 seconds hides a degraded provider behind a slow user experience. The python asyncio wait_for timeout llm api call is a product decision, not just a defensive guard.
Remember that wait_for does not retry. It cancels. If you need retries, wrap safe_complete in a loop with backoff, but cap total attempt time with another outer wait_for so the user-facing deadline holds.
async def complete_with_retry(prompt: str, client: httpx.AsyncClient, attempts: int = 3):
for i in range(attempts):
try:
return await asyncio.wait_for(complete(prompt, client), timeout=2.0)
except asyncio.TimeoutError:
if i == attempts - 1:
raise
await asyncio.sleep(0.1 * (2 ** i))That is the full pattern: network-level timeout in the client, business-level deadline in wait_for, concurrency bound by a semaphore, streaming handled explicitly, and a fault test proving the path works. Ship it.