diff --git a/docs/source/asyncio-client-example.rst b/docs/source/asyncio-client-example.rst
new file mode 100644
index 000000000..26fc78127
--- /dev/null
+++ b/docs/source/asyncio-client-example.rst
@@ -0,0 +1,38 @@
+Asyncio Example Client
+======================
+
+This client sends one HTTPS GET request using :mod:`asyncio` streams. It shows
+how to drive h2 from a coroutine without adding networking or concurrency to
+the protocol implementation itself. It disables server push and collects only
+the requested stream.
+
+Run it with a server that supports HTTP/2 over TLS:
+
+.. code-block:: console
+
+   $ python examples/asyncio/asyncio-client.py nghttp2.org /httpbin/get
+
+The example uses the system trust store, verifies the server certificate and
+hostname, and checks that ALPN negotiated ``h2`` before sending HTTP/2 bytes.
+``--port`` selects a non-default HTTPS port. Headers and trailers go to standard
+error; the response body goes to standard output as bytes.
+
+The same coroutine reads network data, handles h2 events, and flushes generated
+frames with ``write()`` and ``drain()``. Received data is acknowledged using its
+``flow_controlled_length`` (which includes padding), so responses larger than
+the initial receive window can finish. ``StreamEnded`` marks a complete
+response, including responses that end in trailers or contain no body. A reset,
+a premature EOF, or GOAWAY before completion fails the request. The transport
+is closed on success, failure, cancellation, and the command's 15-second timeout.
+
+This is a complete single-request example, not a general-purpose HTTP client.
+It buffers the response body in memory and does not follow redirects, decode
+content encodings, upload a body, retry requests, or multiplex streams. An
+application handling large bodies should consume chunks as they arrive and
+acknowledge them after consumption; one reusing a connection must manage the
+lifetimes of its individual streams as well as the connection.
+
+.. literalinclude:: ../../examples/asyncio/asyncio-client.py
+   :language: python
+   :linenos:
+   :encoding: utf-8
diff --git a/docs/source/examples.rst b/docs/source/examples.rst
index c2f7943f4..4a700c464 100644
--- a/docs/source/examples.rst
+++ b/docs/source/examples.rst
@@ -25,6 +25,7 @@ Example Clients
 .. toctree::
    :maxdepth: 2
 
+   asyncio-client-example
    plain-sockets-example
    twisted-head-example
    twisted-post-example
diff --git a/examples/asyncio/asyncio-client.py b/examples/asyncio/asyncio-client.py
new file mode 100644
index 000000000..912a8b0ea
--- /dev/null
+++ b/examples/asyncio/asyncio-client.py
@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""A single-request HTTP/2 client using asyncio streams and verified TLS."""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import ssl
+import sys
+
+import h2.config
+import h2.connection
+import h2.events
+import h2.settings
+
+
+async def get(host, port=443, path="/", *, context=None):
+    """Return response headers, body bytes and trailers for one HTTPS GET."""
+    if context is None:
+        context = ssl.create_default_context()
+    context.set_alpn_protocols(["h2"])
+    reader, writer = await asyncio.open_connection(
+        host, port, ssl=context, server_hostname=host,
+    )
+    try:
+        if writer.get_extra_info("ssl_object").selected_alpn_protocol() != "h2":
+            raise RuntimeError("The server did not negotiate HTTP/2")
+
+        # One coroutine owns this connection and all calls into its h2 state.
+        conn = h2.connection.H2Connection(
+            config=h2.config.H2Configuration(header_encoding="utf-8"),
+        )
+        conn.initiate_connection()
+        conn.update_settings({h2.settings.SettingCodes.ENABLE_PUSH: 0})
+        stream_id = conn.get_next_available_stream_id()
+        authority = f"[{host}]" if ":" in host else host
+        if port != 443:
+            authority += f":{port}"
+        conn.send_headers(stream_id, [
+            (":method", "GET"), (":scheme", "https"),
+            (":authority", authority), (":path", path),
+        ], end_stream=True)
+        writer.write(conn.data_to_send())
+        await writer.drain()
+
+        headers, trailers = [], []
+        body = bytearray()
+        complete = False
+        terminated = False
+        while not complete:
+            data = await reader.read(65536)
+            if not data:
+                raise EOFError("Connection closed before the response ended")
+            events = conn.receive_data(data)
+            # receive_data processes the whole batch before returning events.
+            # A GOAWAY can accompany the completed response in that batch.
+            terminated = any(isinstance(e, h2.events.ConnectionTerminated) for e in events)
+            for event in events:
+                if isinstance(event, h2.events.DataReceived):
+                    if event.stream_id == stream_id:
+                        body.extend(event.data)
+                    if not terminated:
+                        conn.acknowledge_received_data(
+                            event.flow_controlled_length, event.stream_id,
+                        )
+                elif getattr(event, "stream_id", None) != stream_id:
+                    continue
+                elif isinstance(event, h2.events.ResponseReceived):
+                    headers = event.headers
+                elif isinstance(event, h2.events.TrailersReceived):
+                    trailers = event.headers
+                elif isinstance(event, h2.events.StreamReset):
+                    raise RuntimeError(f"Response stream reset: {event.error_code}")
+                elif isinstance(event, h2.events.StreamEnded):
+                    complete = True
+            if terminated and not complete:
+                raise RuntimeError("Server sent GOAWAY before the response ended")
+            # Flush SETTINGS/PING acknowledgments and WINDOW_UPDATE frames.
+            writer.write(conn.data_to_send())
+            await writer.drain()
+
+        if not terminated:
+            conn.close_connection()
+            writer.write(conn.data_to_send())
+            await writer.drain()
+        return headers, bytes(body), trailers
+    finally:
+        # Also release the transport on errors, cancellation and timeout.
+        writer.close()
+        await writer.wait_closed()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("host")
+    parser.add_argument("path", nargs="?", default="/")
+    parser.add_argument("--port", type=int, default=443)
+    args = parser.parse_args()
+    headers, body, trailers = asyncio.run(asyncio.wait_for(
+        get(args.host, args.port, args.path), timeout=15,
+    ))
+    for name, value in headers + trailers:
+        print(f"{name}: {value}", file=sys.stderr)
+    sys.stdout.buffer.write(body)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_asyncio_client_example.py b/tests/test_asyncio_client_example.py
new file mode 100644
index 000000000..fe32c8a0c
--- /dev/null
+++ b/tests/test_asyncio_client_example.py
@@ -0,0 +1,168 @@
+"""Exercise the documented client against a local TLS HTTP/2 server."""
+from __future__ import annotations
+
+import asyncio
+import importlib.util
+from pathlib import Path
+import shutil
+import ssl
+import subprocess
+
+import pytest
+
+import h2.config
+import h2.connection
+import h2.events
+
+
+@pytest.fixture(scope="module")
+def client():
+    path = Path(__file__).parents[1] / "examples/asyncio/asyncio-client.py"
+    spec = importlib.util.spec_from_file_location("asyncio_client_example", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.fixture(scope="module")
+def tls(tmp_path_factory):
+    openssl = shutil.which("openssl")
+    if openssl is None:
+        pytest.skip("openssl is required to generate a local test certificate")
+    path = tmp_path_factory.mktemp("asyncio-client-tls")
+    cert, key = path / "cert.pem", path / "key.pem"
+    subprocess.run([
+        openssl, "req", "-x509", "-newkey", "rsa:2048", "-nodes",
+        "-keyout", str(key), "-out", str(cert), "-days", "1",
+        "-subj", "/CN=localhost", "-addext", "subjectAltName=DNS:localhost",
+    ], check=True, capture_output=True)
+    server = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    server.load_cert_chain(cert, key)
+    server.set_alpn_protocols(["h2"])
+    trusted = ssl.create_default_context(cafile=str(cert))
+    return server, trusted
+
+
+async def exercise(client, tls, mode, payload=b""):
+    server_context, client_context = tls
+    done = asyncio.get_running_loop().create_future()
+    disconnected = asyncio.Event()
+    requested = asyncio.Event()
+    request = []
+
+    async def respond(reader, writer):
+        conn = h2.connection.H2Connection(config=h2.config.H2Configuration(
+            client_side=False, header_encoding="utf-8",
+        ))
+        conn.initiate_connection()
+        writer.write(conn.data_to_send())
+        offset, stream_id = 0, None
+        sent = False
+        try:
+            while data := await reader.read(65536):
+                for event in conn.receive_data(data):
+                    if isinstance(event, h2.events.RequestReceived):
+                        assert not conn.remote_settings.enable_push
+                        request.extend(event.headers)
+                        requested.set()
+                        stream_id = event.stream_id
+                        if mode == "reset":
+                            conn.reset_stream(stream_id)
+                        elif mode == "goaway":
+                            conn.close_connection()
+                        elif mode == "eof":
+                            return
+                        elif mode != "wait":
+                            conn.send_headers(stream_id, [(":status", "103"), ("link", "</style.css>")])
+                            conn.send_headers(stream_id, [(":status", "200")], end_stream=mode == "empty")
+                            sent = mode == "empty"
+                    elif isinstance(event, h2.events.ConnectionTerminated):
+                        return
+                if stream_id and not sent and mode in {"body", "trailers", "complete-goaway"}:
+                    while offset < len(payload) and conn.local_flow_control_window(stream_id) > 1:
+                        # Count the pad-length byte too, to test flow_controlled_length.
+                        size = min(len(payload) - offset, conn.local_flow_control_window(stream_id) - 1,
+                                   conn.max_outbound_frame_size - 1)
+                        conn.send_data(stream_id, payload[offset:offset + size], pad_length=0)
+                        offset += size
+                    if offset == len(payload):
+                        if mode == "trailers":
+                            conn.send_headers(stream_id, [("x-finished", "yes")], end_stream=True)
+                        else:
+                            conn.end_stream(stream_id)
+                        if mode == "complete-goaway":
+                            conn.close_connection()
+                        sent = True
+                writer.write(conn.data_to_send())
+                await writer.drain()
+        except Exception as exc:
+            if not done.done():
+                done.set_exception(exc)
+        finally:
+            writer.close()
+            await writer.wait_closed()
+            disconnected.set()
+            if not done.done():
+                done.set_result(None)
+
+    server = await asyncio.start_server(respond, "127.0.0.1", 0, ssl=server_context)
+    port = server.sockets[0].getsockname()[1]
+    async with server:
+        task = asyncio.create_task(client.get("localhost", port, "/resource?q=1", context=client_context))
+        try:
+            if mode == "wait":
+                await asyncio.wait_for(requested.wait(), 5)
+                with pytest.raises(asyncio.TimeoutError):
+                    await asyncio.wait_for(task, 0.2)
+                result = None
+            elif mode in {"reset", "goaway", "eof"}:
+                with pytest.raises((RuntimeError, EOFError)):
+                    await asyncio.wait_for(task, 5)
+                result = None
+            else:
+                result = await asyncio.wait_for(task, 5)
+            await asyncio.wait_for(disconnected.wait(), 5)
+            await done
+            assert (":authority", f"localhost:{port}") in request
+            assert (":path", "/resource?q=1") in request
+            return result
+        finally:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+
+
+@pytest.mark.parametrize("mode", ["body", "trailers", "empty", "complete-goaway"])
+def test_response(client, tls, mode):
+    payload = bytes(range(256)) * 1024 if mode != "empty" else b""
+    headers, body, trailers = asyncio.run(exercise(client, tls, mode, payload))
+    assert headers == [(":status", "200")]
+    assert body == payload
+    assert trailers == ([("x-finished", "yes")] if mode == "trailers" else [])
+
+
+@pytest.mark.parametrize("mode", ["reset", "goaway", "eof", "wait"])
+def test_failure_closes_transport(client, tls, mode):
+    asyncio.run(exercise(client, tls, mode))
+
+
+def test_alpn_failure_sends_no_h2_bytes(client, tls):
+    async def check():
+        server_context, trusted = tls
+        server_context.set_alpn_protocols(["http/1.1"])
+        received = asyncio.get_running_loop().create_future()
+
+        async def respond(reader, writer):
+            received.set_result(await reader.read())
+            writer.close()
+            await writer.wait_closed()
+
+        try:
+            server = await asyncio.start_server(respond, "127.0.0.1", 0, ssl=server_context)
+            async with server:
+                with pytest.raises(RuntimeError, match="negotiate HTTP/2"):
+                    await client.get("localhost", server.sockets[0].getsockname()[1], context=trusted)
+                assert await asyncio.wait_for(received, 5) == b""
+        finally:
+            server_context.set_alpn_protocols(["h2"])
+
+    asyncio.run(check())