Asyncio Streams, Transports & Protocols

Below every async HTTP client, database driver and message broker library sits the same small set of primitives: a transport that owns the socket, a protocol object that receives bytes as they arrive, and — for code that prefers await to callbacks — the StreamReader/StreamWriter pair layered on top. Most application code never touches them. But the moment you have to speak a wire protocol nobody has packaged (a device on a factory network, a legacy TCP service, a line-oriented control channel), these are the tools, and the failure modes they expose are unlike anything in the request/response world: partial reads, frames split across packets, writers that accept data faster than the socket can drain, and connections that stay half-open for hours.

This section covers building on them safely. It maps the streams API onto the callback-based transport/protocol layer beneath it, works through framing (the single most common source of protocol bugs), explains why await writer.drain() is not optional, sizes the reader's buffer limits, and lays out connection teardown that does not lose the last write. The parent overview, Async Network I/O & Protocol Handling, covers the ready-made clients; this section is for when there is no client to reach for.

Scope of this section:

• The streams API (open_connection, start_server) and the transport/protocol layer underneath it.
• Framing strategies: delimiters, fixed headers, length prefixes — and why read() is never enough.
• Backpressure: what drain() actually waits for, and the high/low water marks behind it.
• Buffer limits, readuntil() overflow, and defending against a peer that never sends a delimiter.
• Half-close, EOF handling, and shutting a connection down without truncating in-flight data.

Architectural principles

• TCP is a byte stream, not a message stream. Nothing preserves your write boundaries. A 400-byte message may arrive as one read, or as 40 plus 360, or glued to the next message. Every correct protocol carries its own framing.
• Every read needs a bound. read() without a limit, readuntil() without limit=, or a frame length taken from the wire without validation are all ways for a peer — malicious or merely broken — to allocate your memory for you.
• Writes are buffered; drain() is the flow control. writer.write() never blocks. Without an accompanying await writer.drain(), a fast producer and a slow consumer fill an unbounded buffer inside your own process.
• Protocol callbacks must not block or await. data_received() runs on the loop thread; heavy parsing there stalls every other connection. Push work into a queue or a task and return quickly.
• Streams are convenience, transports are control. Streams are the right default. Drop to the protocol layer only when you need per-datagram control, custom flow-control thresholds, or to avoid the task-per-connection model.

Execution model: from selector to your coroutine

When a byte arrives, the selector marks the socket readable and the loop calls the transport's read-ready callback. The transport does one recv(), then calls protocol.data_received(data) synchronously. For stream-based code, that protocol is asyncio's own StreamReaderProtocol, which feeds the bytes into a StreamReader buffer and wakes any coroutine parked on read()/readline()/readexactly() by resolving its future — which schedules your coroutine to resume on the next loop iteration. Your await returns one hop later, with whatever the buffer now holds.

Three consequences follow. First, await reader.read(1024) means "give me up to 1024 bytes, as soon as at least one is available" — it is not a promise of 1024 bytes, which is why readexactly() exists. Second, the parsing you do after that await runs on the loop thread, so a slow parser is a slow event loop, exactly the stall covered in event loop debugging and instrumentation. Third, flow control is a property of the transport: when the reader's buffer exceeds its high-water mark, the transport calls pause_reading() and stops polling that socket, so TCP's own window closes and the peer is throttled — backpressure propagating all the way to the sender without your code doing anything, provided you sized the limits sensibly.

Writing works in reverse. writer.write(data) appends to the transport's send buffer and returns immediately; the loop drains it when the socket is writable. If the buffer exceeds the write high-water mark (64 KiB by default), the transport signals pause_writing(), and that is the state await writer.drain() waits to clear. Skip the drain() and the buffer simply grows — the writer never refuses data.

Pattern catalogue

Line-delimited protocol with the streams API

The simplest framing, and the right choice for text control protocols.

import asyncio


async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
    peer = writer.get_extra_info("peername")
    try:
        while True:
            line = await reader.readline()          # bounded by the reader's limit
            if not line:                            # EOF: peer closed cleanly
                break
            response = process(line.rstrip(b"\r\n"))
            writer.write(response + b"\n")
            await writer.drain()                    # flow control, not optional
    except asyncio.IncompleteReadError:
        pass                                        # peer vanished mid-line
    finally:
        writer.close()
        await writer.wait_closed()


async def main() -> None:
    server = await asyncio.start_server(handle, "0.0.0.0", 9000, limit=64 * 1024)
    async with server:
        await server.serve_forever()

readline() raises LimitOverrunError if the delimiter never arrives within the reader's limit, which is the behaviour you want: a peer that sends 4 GB without a newline gets an error rather than your memory. Line framing breaks down as soon as payloads may contain the delimiter — at which point you need escaping or a different frame.

Length-prefixed framing

The standard for binary protocols: a fixed-size header carries the payload length.

import asyncio
import struct

HEADER = struct.Struct("!I")            # 4-byte big-endian length
MAX_FRAME = 8 * 1024 * 1024             # refuse anything larger


async def read_frame(reader: asyncio.StreamReader) -> bytes:
    header = await reader.readexactly(HEADER.size)
    (length,) = HEADER.unpack(header)
    if length > MAX_FRAME:              # never trust a length from the wire
        raise ValueError(f"frame of {length} bytes exceeds the {MAX_FRAME} limit")
    return await reader.readexactly(length)


async def write_frame(writer: asyncio.StreamWriter, payload: bytes) -> None:
    writer.write(HEADER.pack(len(payload)) + payload)     # one write: never split
    await writer.drain()

readexactly() is what makes this correct — it loops internally until the full count arrives and raises IncompleteReadError on a short stream. Writing the header and payload in a single write() avoids interleaving when two tasks share a writer, which is otherwise a subtle corruption bug.

The transport/protocol layer

Drop below streams when you want explicit control of flow-control thresholds or a connection model without a task per connection.

import asyncio


class LineProtocol(asyncio.Protocol):
    def connection_made(self, transport: asyncio.Transport) -> None:
        self.transport = transport
        self.buffer = bytearray()
        transport.set_write_buffer_limits(high=256 * 1024, low=64 * 1024)

    def data_received(self, data: bytes) -> None:
        self.buffer.extend(data)
        while (idx := self.buffer.find(b"\n")) != -1:
            line, self.buffer = bytes(self.buffer[:idx]), self.buffer[idx + 1:]
            self.transport.write(process(line) + b"\n")   # never await in here
        if len(self.buffer) > 1_000_000:                  # no delimiter in sight
            self.transport.abort()

    def pause_writing(self) -> None:      # send buffer above the high-water mark
        self.paused = True

    def resume_writing(self) -> None:
        self.paused = False

    def connection_lost(self, exc: Exception | None) -> None:
        release_resources(self)

The rule that makes this layer safe is that no callback may block or await; anything expensive goes onto a queue that a separate task drains. Note that the buffer bound has to be enforced by you — the transport does not know what a frame is.

Client connection with timeouts

An open_connection() without a timeout will happily wait for a peer that will never answer.

import asyncio


async def connect(host: str, port: int, budget: float = 5.0):
    async with asyncio.timeout(budget):                    # covers DNS + TCP + TLS
        reader, writer = await asyncio.open_connection(host, port, limit=128 * 1024)
    return reader, writer

Wrap the connect, every read, and the shutdown in their own deadlines — the reasoning is in choosing asyncio.timeout vs wait_for. A read without a deadline is the most common cause of a task that never returns.

Clean shutdown, including half-close

Closing correctly is a two-step process, and skipping the second step truncates data.

async def shutdown(writer: asyncio.StreamWriter) -> None:
    if writer.can_write_eof():
        writer.write_eof()              # half-close: we are done sending
    writer.close()                      # start closing the transport
    try:
        async with asyncio.timeout(5):
            await writer.wait_closed()  # flush pending writes, then release the fd
    except TimeoutError:
        writer.transport.abort()        # peer is not draining: drop it

close() alone returns immediately and the flush happens in the background — if the process exits first, the last frames are lost. wait_closed() is what makes shutdown deterministic, and the timeout around it is what stops a stuck peer from holding your shutdown hostage.

Resource boundaries

Every knob here is a memory bound, and the defaults are chosen for a general case that may not be yours.

Limit	Default	Effect when too small	Effect when too large
limit= on open_connection/start_server	64 KiB	LimitOverrunError on legitimate frames	A single peer can pin limit bytes per connection
Write high-water mark	64 KiB	drain() returns constantly; more wakeups	Larger latency spike and more memory per slow peer
Write low-water mark	high ÷ 4	Frequent pause/resume churn	Bursty draining
Application frame cap (MAX_FRAME)	none — you set it	Valid messages rejected	Memory exhaustion from a hostile length prefix
Concurrent connections	unbounded	Rejected clients	fd exhaustion; cap it with a semaphore

The arithmetic that matters: per-connection worst-case memory is roughly the read limit plus the write high-water mark plus your own frame buffer. At 10,000 connections with defaults, that is well over a gigabyte before any application data — which is why servers that hold many idle connections lower these limits rather than raising them.

Integrated production example

A length-prefixed request/response server with bounded frames, per-connection concurrency limits, deadlines on every phase, backpressure on writes, and metrics.

import asyncio
import logging
import struct
import time

log = logging.getLogger("proto")
HEADER = struct.Struct("!I")
MAX_FRAME = 4 * 1024 * 1024
IDLE_TIMEOUT = 60.0
READ_TIMEOUT = 30.0

STATS = {"connections": 0, "frames": 0, "overlong": 0, "idle_closed": 0}


async def read_frame(reader: asyncio.StreamReader) -> bytes:
    header = await reader.readexactly(HEADER.size)
    (length,) = HEADER.unpack(header)
    if length > MAX_FRAME:
        STATS["overlong"] += 1
        raise ValueError(f"frame {length} > {MAX_FRAME}")
    return await reader.readexactly(length)


async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
    peer = writer.get_extra_info("peername")
    STATS["connections"] += 1
    started = time.monotonic()
    try:
        while True:
            try:
                async with asyncio.timeout(IDLE_TIMEOUT):     # idle peers are closed
                    payload = await read_frame(reader)
            except TimeoutError:
                STATS["idle_closed"] += 1
                log.info("closing idle connection from %s", peer)
                break
            except (asyncio.IncompleteReadError, ConnectionResetError):
                break                                          # peer went away

            STATS["frames"] += 1
            try:
                async with asyncio.timeout(READ_TIMEOUT):
                    response = await handle_request(payload)
            except TimeoutError:
                response = b'{"error":"timeout"}'

            writer.write(HEADER.pack(len(response)) + response)
            await writer.drain()                               # backpressure here
    except ValueError as exc:
        log.warning("protocol error from %s: %s", peer, exc)
    except Exception:
        log.exception("unexpected failure serving %s", peer)
    finally:
        writer.close()
        try:
            async with asyncio.timeout(5):
                await writer.wait_closed()
        except (TimeoutError, ConnectionResetError):
            writer.transport.abort()
        log.info("connection from %s closed after %.1fs", peer, time.monotonic() - started)


async def handle_request(payload: bytes) -> bytes:
    await asyncio.sleep(0)                                     # your logic here
    return b'{"ok":true}'


async def monitor(server: asyncio.Server, interval: float = 30.0) -> None:
    while True:
        await asyncio.sleep(interval)
        live = len(getattr(server, "_active", ()) or ())
        log.info("frames=%d overlong=%d idle_closed=%d live=%d",
                 STATS["frames"], STATS["overlong"], STATS["idle_closed"], live)


async def main() -> None:
    server = await asyncio.start_server(handle, "0.0.0.0", 9000, limit=256 * 1024)
    async with server, asyncio.TaskGroup() as tg:
        tg.create_task(monitor(server), name="proto.monitor")
        await server.serve_forever()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

Three details are what make this production code rather than an example. Every phase has its own deadline, so a peer can neither hold a connection open forever nor stall mid-request. The frame cap is enforced before readexactly() allocates, which is the difference between rejecting a hostile 4 GB length and dying from it. And the teardown path has both a graceful branch and an abort() fallback, so a peer that refuses to drain cannot block shutdown.

Diagnostic Hook — the four numbers a custom protocol needs

Export live connections (fd pressure), frames per second (throughput), drain wait time (a histogram of seconds inside await writer.drain()), and protocol errors by kind (overlong frame, incomplete read, idle timeout). Drain wait is the one people forget and the one that explains the most: near zero means consumers keep up; consistently non-zero means a slow peer is now applying backpressure to your producer, and if that producer is a broadcast loop, one slow client is throttling everyone — the failure analysed in handling WebSocket backpressure with slow consumers. Alert on live connections approaching your fd limit and on any sustained rise in overlong-frame errors, which usually means a version mismatch rather than an attack.

Failure modes

Failure mode	Root cause	Detection	Fix
Messages merge or split	Reading with read(n) and assuming message boundaries	Parser sees truncated or concatenated payloads	Frame explicitly: readexactly() with a length prefix, or readline()
Unbounded memory on one connection	read() with no limit, or a trusted length prefix	RSS tracks one peer's send rate	Set limit=, enforce MAX_FRAME before reading
LimitOverrunError on valid traffic	Reader limit smaller than the largest legitimate frame	Errors correlate with large messages	Raise limit= to just above the real maximum
Send buffer grows without bound	write() without await drain()	Memory climbs on the writer; slow peer	Always drain(); add a per-peer send queue with a bound
Last response lost on shutdown	close() without await wait_closed()	Clients see truncated final frames	Await wait_closed() with a timeout, then abort()
Loop stalls under load	Parsing inside data_received() or a CPU-heavy parse after await	Slow-callback warnings; lag p99 spikes	Queue the payload; parse in a worker or offload
Connections leak	Handler raises before finally; no idle timeout	Live-connection gauge climbs monotonically	Close in finally; enforce an idle deadline

Frequently Asked Questions

When should I use asyncio streams instead of the transport and protocol API?

Use streams by default: open_connection and start_server give you awaitable reads, automatic flow control through drain(), and a task per connection, which is far easier to reason about. Drop to the transport/protocol layer when you need explicit control over write buffer limits, want to avoid one task per connection at very high connection counts, or are implementing UDP, where there is no stream abstraction.

Why do I need await writer.drain() after writer.write()?

Because write() never blocks — it appends to the transport's send buffer and returns. If the peer reads more slowly than you write, that buffer grows without limit inside your process. drain() waits until the buffer falls back below the low-water mark, which is how backpressure reaches your producer. Without it, a single slow consumer becomes an out-of-memory failure.

How do I frame messages over TCP in asyncio?

Either delimit them (readline() or readuntil() with an explicit limit) or prefix each payload with a fixed-size length header and use readexactly() for the header and then the body. Always validate the decoded length against a maximum before reading, because the length arrives from the peer and an unvalidated value is a memory-exhaustion vector.

What does the limit parameter on open_connection actually control?

It is the maximum number of bytes StreamReader will buffer while waiting for a delimiter or requested count — 64 KiB by default. Exceeding it raises LimitOverrunError for readuntil()/readline(). Size it just above your largest legitimate frame: too small and valid messages fail, too large and each connected peer can pin that much memory.

How do I close an asyncio connection without losing buffered data?

Call writer.write_eof() if you are done sending and the transport supports it, then writer.close(), then await writer.wait_closed() so pending writes flush before the socket is released. Wrap the wait in a timeout and fall back to transport.abort(), otherwise a peer that stops reading can hold your shutdown open indefinitely.

Related

• Async Network I/O & Protocol Handling — up to the overview for the transport mental model and the ready-made clients above this layer.
• Reading line-delimited protocols with asyncio streams — readline, readuntil, and defending against a missing delimiter.
• Implementing a length-prefixed framing protocol — a complete binary frame codec with validation and tests.
• Handling drain and write backpressure in asyncio streams — water marks, per-peer queues, and slow-consumer policy.
• Connection Pooling & Keep-Alive — reusing the connections this layer creates.