When a user closes a tab or a downstream step times out, you need to python asyncio cancel streaming request work that is mid-flight. Leaving the stream running wastes tokens, holds sockets, and can wedge your event loop with orphaned tasks. The patterns below show exactly how to tear down an async LLM stream using tasks, asyncio.wait_for, and proper context managers, with runnable code you can drop into a service.
Step 1: Pick a client that exposes async streaming
Use either the official openai async client or a raw httpx client against any OpenAI-compatible endpoint. If you’re hitting multiple providers, a gateway like n4n.ai gives one OpenAI-compatible endpoint for 240+ models with automatic fallback; the cancellation logic below is identical because the wire protocol doesn’t change.
For explicit control, httpx is easiest to reason about:
import httpx, json, asyncio
API_URL = "https://api.openai.com/v1/chat/completions"
HEADERS = {"Authorization": "Bearer $API_KEY", "Content-Type": "application/json"}
async def stream_llm(messages, model="gpt-4o-mini"):
payload = {"model": model, "messages": messages, "stream": True}
async with httpx.AsyncClient() as client:
async with client.stream("POST", API_URL, json=payload, headers=HEADERS) as resp:
async for line in resp.aiter_lines():
if line.startswith("data: "):
data = line[len("data: "):].strip()
if data == "[DONE]":
break
yield json.loads(data)The async with blocks guarantee the connection is released even if the consumer is cancelled.
Step 2: Run the consumer inside a task
Never iterate the stream at the top level of your handler if you want to cancel it externally. Wrap the consumption in a coroutine and schedule it with asyncio.create_task.
async def consume(stream, queue: asyncio.Queue):
async for chunk in stream:
await queue.put(chunk["choices"][0]["delta"].get("content", ""))
# simulate slow worker
await asyncio.sleep(0.01)
async def main():
messages = [{"role": "user", "content": "Write a long poem."}]
stream = stream_llm(messages)
queue: asyncio.Queue = asyncio.Queue()
consumer = asyncio.create_task(consume(stream, queue))
# let it run briefly
await asyncio.sleep(0.5)
consumer.cancel()
try:
await consumer
except asyncio.CancelledError:
print("consumer cancelled")Cancelling consumer raises CancelledError at the next await point inside consume—usually the async for yield or the queue.put. Because stream_llm uses async with, the generator’s aclose is invoked and the TCP connection is closed.
Step 3: Cancel from a signal or external event
In a real server you cancel in response to a client disconnect or a shutdown signal. Here is a minimal aiohttp handler pattern:
from aiohttp import web
async def handle(request):
messages = [{"role": "user", "content": request.query.get("q", "")}]
stream = stream_llm(messages)
queue: asyncio.Queue = asyncio.Queue()
consumer = asyncio.create_task(consume(stream, queue))
request.app["active"].add(consumer)
try:
while True:
if request.transport.is_closing():
break
try:
chunk = await asyncio.wait_for(queue.get(), timeout=0.1)
await request.write(chunk.encode())
except asyncio.TimeoutError:
continue
finally:
consumer.cancel()
await request.app["active"].discard(consumer)
try:
await consumer
except asyncio.CancelledError:
pass
return web.Response(text="stream ended")The finally block is the critical part: it ensures the python asyncio cancel streaming request task even if the client hangs up. Without it, the task leaks.
Step 4: Use asyncio.wait_for for timeout-based cancellation
If you only need a hard deadline, skip manual cancel() and use asyncio.wait_for. It cancels the inner coroutine when the timeout elapses.
async def bounded_stream(messages, timeout=2.0):
stream = stream_llm(messages)
queue: asyncio.Queue = asyncio.Queue()
try:
await asyncio.wait_for(consume(stream, queue), timeout)
except asyncio.TimeoutError:
print("stream timed out, task cancelled automatically")
# stream's async with still cleans upwait_for cancels the wrapped task and waits for it to observe the cancellation. The same context-manager cleanup applies.
Step 5: Handle CancelledError in the producer
If your streaming generator does non-trivial setup (e.g., retrying on partial failure), catch CancelledError to log or metrics, but always re-raise.
async def stream_llm_safe(messages):
try:
async for chunk in stream_llm(messages):
yield chunk
except asyncio.CancelledError:
# record metric: cancelled mid-stream
raiseSwallowing CancelledError will break asyncio’s cancellation contract and hang the task.
Step 6: Verify the cancellation worked
A cancelled stream should leave no pending tasks and no open connections. Add a verification routine:
async def verify_clean_shutdown():
before = len(asyncio.all_tasks())
messages = [{"role": "user", "content": "Count to 1000."}]
stream = stream_llm(messages)
queue: asyncio.Queue = asyncio.Queue()
task = asyncio.create_task(consume(stream, queue))
await asyncio.sleep(0.2)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
await asyncio.sleep(0.1)
after = len(asyncio.all_tasks())
assert after == before, f"leaked {after - before} tasks"
print("OK: no leaked tasks, stream closed")Run it with asyncio.run(verify_clean_shutdown()). If the assertion holds and you see OK, the python asyncio cancel streaming request path is solid. For socket verification, inspect lsof -i or netstat before and after; the ESTABLISHED connection to the LLM host should disappear within a second of cancellation.
Edge cases that bite in production
Generator close ordering
If you manually drive an async generator (not via async for), you must call await gen.aclose(). Using async for or async with does this for you. Never write chunk = await stream.__anext__() in a loop without a finally: await stream.aclose().
Shielded cleanup
If you need to run final logging after cancel, wrap only the logging in asyncio.shield, not the stream consumption:
async def consume_with_logging(stream, queue):
try:
async for chunk in stream:
await queue.put(chunk)
except asyncio.CancelledError:
await asyncio.shield(log_cancel()) # fire but don't block cancel
raiseMultiple consumers
If you fan out one stream to many websockets, cancel the producer when all consumers are gone. Track reference counts; cancel the producer task only after the last release().
Minimal runnable script
import asyncio, httpx, json
API_URL = "https://api.openai.com/v1/chat/completions"
HEADERS = {"Authorization": "Bearer $API_KEY", "Content-Type": "application/json"}
async def stream_llm(messages):
async with httpx.AsyncClient() as client:
async with client.stream("POST", API_URL,
json={"model": "gpt-4o-mini", "messages": messages, "stream": True},
headers=HEADERS) as resp:
async for line in resp.aiter_lines():
if line.startswith("data: "):
data = line[6:].strip()
if data == "[DONE]":
break
yield json.loads(data)
async def consume(stream, q: asyncio.Queue):
async for c in stream:
await q.put(c["choices"][0]["delta"].get("content", ""))
async def main():
stream = stream_llm([{"role": "user", "content": "Tell me a story."}])
q: asyncio.Queue = asyncio.Queue()
t = asyncio.create_task(consume(stream, q))
await asyncio.sleep(0.3)
t.cancel()
try:
await t
except asyncio.CancelledError:
print("cancelled cleanly")
print("remaining tasks:", len(asyncio.all_tasks()) - 1)
if __name__ == "__main__":
asyncio.run(main())Replace $API_KEY and run. You should see cancelled cleanly and remaining tasks: 0. That confirms the python asyncio cancel streaming request pattern holds end to end.