Drop is_valid_h2c_settings, validate via the real upgrade path instead

The pre-flight validator was running every payload through a throwaway
H2Connection just to ask "would h2 accept this?" before doing the same
work on the real connection. That's redundant: hyper-h2 raises out of
`initiate_upgrade_connection` for malformed payloads on its own.

Make `H2Protocol.initiate_h2c_upgrade` return a bool. The HTTP/1.1
caller now attempts the upgrade up front and only commits the
`101 Switching Protocols` response if hyper-h2 accepts the SETTINGS
payload. When hyper-h2 rejects it the H1 protocol replies with 400
instead of leaving the wire in a half-broken state.

docs/concepts/http2.md

@@ -0,0 +1,241 @@
+**Uvicorn** supports HTTP/2, the major revision of the HTTP protocol that provides significant
+performance improvements over HTTP/1.1.
+
+!!! warning "Experimental Feature"
+    HTTP/2 support is currently **experimental** and is **not enabled by default**.
+
+## Overview
+
+HTTP/2 introduces several key features:
+
+- **Multiplexing**: Multiple requests and responses can be sent simultaneously over a single TCP connection
+- **Header compression**: HTTP headers are compressed using HPACK, reducing overhead
+- **Binary protocol**: More efficient parsing compared to HTTP/1.1's text-based format
+- **Stream prioritization**: Clients can indicate which resources are more important
+
+## Enabling HTTP/2
+
+To enable HTTP/2 support in Uvicorn, use the `--http2` flag:
+
+=== "Command Line"
+    ```bash
+    uvicorn main:app --http2
+    ```
+
+=== "Programmatic"
+    ```python
+    import uvicorn
+
+    uvicorn.run("main:app", http2=True)
+    ```
+
+!!! note
+    HTTP/2 support requires the `h2` package. Install it with:
+    ```bash
+    pip install h2
+    ```
+
+## Connection Methods
+
+HTTP/2 can be established through two different mechanisms: **h2** (over TLS) and **h2c** (cleartext).
+
+### h2: HTTP/2 over TLS (Recommended)
+
+When using HTTPS, HTTP/2 is negotiated via **ALPN** (Application-Layer Protocol Negotiation)
+during the TLS handshake. This is the most common and recommended way to use HTTP/2.
+
+```mermaid
+sequenceDiagram
+    participant Client
+    participant Server
+
+    Note over Client,Server: TLS Handshake with ALPN
+
+    Client->>Server: ClientHello
+    Note right of Client: ALPN: h2, http/1.1
+
+    Server->>Client: ServerHello
+    Note right of Server: ALPN: h2
+
+    Note over Client,Server: TLS Handshake Complete
+
+    Client->>Server: HTTP/2 Connection Preface
+    Server->>Client: HTTP/2 SETTINGS Frame
+
+    Note over Client,Server: HTTP/2 Connection Established
+
+    Client->>Server: HEADERS (Stream 1)
+    Server->>Client: HEADERS + DATA (Stream 1)
+```
+
+For testing it locally, you can generate a self-signed certificate and use it to test the HTTP/2 connection.
+
+```bash
+openssl req -x509 -newkey rsa:2048 -keyout key.pem -out cert.pem -days 365 -nodes -subj "/CN=localhost"
+```
+
+Then create a simple ASGI application to test the connection.
+
+```python title="main.py"
+async def app(scope, receive, send):
+    await send({"type": "http.response.start", "status": 200, "headers": []})
+    await send({"type": "http.response.body", "body": b"ok"})
+```
+
+Run Uvicorn with the `--http2` flag and the SSL certificate files.
+
+=== "Command Line"
+    ```bash
+    uvicorn app:app --http2 --ssl-keyfile key.pem --ssl-certfile cert.pem
+    ```
+
+=== "Programmatic"
+    ```python
+    import uvicorn
+
+    uvicorn.run(
+        "app:app",
+        http2=True,
+        ssl_keyfile="key.pem",
+        ssl_certfile="cert.pem",
+    )
+    ```
+
+You can test the connection using curl with the `--http2` flag.
+
+```bash
+# Use -k to skip certificate verification for self-signed certs
+curl -v --http2 -k https://localhost:8000/
+```
+
+### h2c: HTTP/2 Cleartext
+
+HTTP/2 can also be used without TLS through an **upgrade mechanism**. The client sends an
+HTTP/1.1 request with upgrade headers, and if the server supports HTTP/2, it responds with
+`101 Switching Protocols`.
+
+```mermaid
+sequenceDiagram
+    participant Client
+    participant Server
+
+    Note over Client,Server: h2c Upgrade Process
+
+    Client->>Server: HTTP/1.1 GET /
+    Note right of Client: Headers:<br/>Upgrade: h2c<br/>HTTP2-Settings: [base64]<br/>Connection: Upgrade, HTTP2-Settings
+
+    Server->>Client: HTTP/1.1 101 Switching Protocols
+    Note right of Server: Headers:<br/>Upgrade: h2c<br/>Connection: Upgrade
+
+    Note over Client,Server: Connection Upgraded to HTTP/2
+
+    Server->>Client: HTTP/2 SETTINGS Frame
+    Client->>Server: HTTP/2 SETTINGS ACK
+
+    Server->>Client: HEADERS + DATA (Stream 1)
+    Note right of Server: Response to original request
+```
+
+Using the same `main.py` from the h2 section above, run Uvicorn with the `--http2` flag.
+
+=== "Command Line"
+    ```bash
+    uvicorn main:app --http2
+    ```
+
+=== "Programmatic"
+    ```python
+    import uvicorn
+
+    uvicorn.run("main:app", http2=True)
+    ```
+
+You can test the connection using curl with the `--http2` flag.
+
+```bash
+curl -v --http2 http://localhost:8000/
+```
+
+!!! warning
+    h2c is not supported by web browsers. Browsers only support HTTP/2 over TLS (h2).
+    h2c is primarily useful for internal services, proxies, or testing.
+
+## ASGI Scope
+
+When a request comes in over HTTP/2, the ASGI scope will have `http_version` set to `"2"`:
+
+```python
+async def app(scope, receive, send):
+    assert scope["type"] == "http"
+    print(f"HTTP Version: {scope['http_version']}")  # "2" for HTTP/2
+    # ... handle request
+```
+
+## Using with Reverse Proxies
+
+In production, Uvicorn is typically deployed behind a reverse proxy like Nginx, Caddy, or HAProxy.
+
+**Benefits of using a reverse proxy:**
+
+- **TLS termination**: The proxy handles SSL/TLS encryption, offloading this work from your application
+- **Load balancing**: Distribute requests across multiple Uvicorn instances
+- **Static file serving**: Serve static assets directly without hitting your Python application
+- **Request buffering**: Buffer slow clients to free up Uvicorn workers
+- **Security**: Hide your application server details, add rate limiting, and filter malicious requests
+- **HTTP/2 to clients**: Provide HTTP/2 benefits to clients even if using HTTP/1.1 internally
+
+```mermaid
+flowchart LR
+    Client <-->|HTTP/2 over TLS| Proxy
+    Proxy <-->|HTTP/1.1 or HTTP/2| Uvicorn
+
+    style Client fill:#e1f5fe
+    style Proxy fill:#fff3e0
+    style Uvicorn fill:#e8f5e9
+```
+
+### Proxy HTTP/2 Upstream Support
+
+**HTTP/2 Upstream** refers to the protocol used between the proxy and the backend server (Uvicorn).
+While all modern proxies support HTTP/2 for client connections, support for HTTP/2 to backend
+servers varies.
+
+**Multiplexing** is HTTP/2's ability to send multiple requests simultaneously over a single TCP
+connection. Without multiplexing, each request requires its own connection, negating a key
+benefit of HTTP/2. Some proxies support HTTP/2 upstream but open a new connection per request,
+which means they don't truly multiplex.
+
+Here's the current state of proxy support (as of 2026-02-02):
+
+| Proxy | HTTP/2 Upstream | Multiplexing | Documentation |
+|-------|-----------------|--------------|---------------|
+| **Envoy** | Yes | Yes | [Connection Pooling Docs](https://www.envoyproxy.io/docs/envoy/latest/intro/arch_overview/upstream/connection_pooling) |
+| **Caddy** | Yes | Yes | [reverse_proxy Docs](https://caddyserver.com/docs/caddyfile/directives/reverse_proxy) |
+| **HAProxy** | Yes | Yes | [HTTP/2 Docs](https://www.haproxy.com/documentation/hapee/latest/load-balancing/protocols/http-2/) |
+| **Traefik** | Yes | Yes | [ServersTransport Docs](https://doc.traefik.io/traefik/routing/services/) |
+| **Apache** | Partial | No | [mod_proxy_http2 Docs](https://httpd.apache.org/docs/trunk/mod/mod_proxy_http2.html) |
+| **Nginx** | Limited | No | [Trac Ticket #923](https://trac.nginx.org/nginx/ticket/923) |
+
+### Recommended Proxy Configuration
+
+For most production deployments, using **HTTP/1.1 with keepalive** connections between the proxy
+and Uvicorn is recommended. This provides excellent performance while being simple to configure
+and debug.
+
+!!! note "h2c Prior Knowledge Not Supported"
+    Uvicorn's h2c implementation uses the HTTP/1.1 upgrade mechanism. It does **not** support
+    "prior knowledge" h2c where clients send the HTTP/2 connection preface directly. This means
+    proxy configurations using `h2c://` URLs will not work.
+
+    For HTTP/2 between proxy and Uvicorn, use **h2 over TLS** (ALPN negotiation).
+
+## Performance Considerations
+
+HTTP/2 provides the most benefit when:
+
+- **High latency connections**: Multiplexing reduces round-trip overhead
+- **Many concurrent requests**: Multiple streams share a single connection
+- **Large headers**: HPACK compression reduces header overhead
+
+For internal, low-latency connections (like proxy to backend), HTTP/1.1 with keepalive
+often performs comparably to HTTP/2, which is why nginx's approach is still effective.

docs/index.md

@@ -42,7 +42,7 @@ Until recently Python has lacked a minimal low-level server/application interfac
 async frameworks. The [ASGI specification](https://asgi.readthedocs.io/en/latest/) fills this gap,
 and means we're now able to start building a common set of tooling usable across all async frameworks.
 
-Uvicorn currently supports **HTTP/1.1** and **WebSockets**.
+Uvicorn currently supports **HTTP/1.1**, **HTTP/2**, and **WebSockets**.
 
 ## Quickstart
 

docs/settings.md

@@ -92,6 +92,7 @@ Using Uvicorn with watchfiles will enable the following options (which are other
 
 * `--loop <str>` - Set the event loop implementation. The uvloop implementation provides greater performance, but is not compatible with Windows or PyPy. **Options:** *'auto', 'asyncio', 'uvloop'.* **Default:** *'auto'*.
 * `--http <str>` - Set the HTTP protocol implementation. The httptools implementation provides greater performance, but it not compatible with PyPy. **Options:** *'auto', 'h11', 'httptools'.* **Default:** *'auto'*.
+* `--http2` - Enable HTTP/2 support. Requires the `h2` package (`pip install h2`). When enabled, HTTP/2 is available via ALPN negotiation over TLS (h2) and via cleartext upgrade (h2c). See the [HTTP/2 documentation](concepts/http2.md) for details. **Default:** *False*.
 * `--ws <str>` - Set the WebSockets protocol implementation. Either of the `websockets` and `wsproto` packages are supported. There are two versions of `websockets` supported: `websockets` and `websockets-sansio`. Use `'none'` to ignore all websocket requests. **Options:** *'auto', 'none', 'websockets', 'websockets-sansio', 'wsproto'.* **Default:** *'auto'*.
 * `--ws-max-size <int>` - Set the WebSockets max message size, in bytes. Only available with the `websockets` protocol. **Default:** *16777216* (16 MB).
 * `--ws-max-queue <int>` - Set the maximum length of the WebSocket incoming message queue. Only available with the `websockets` protocol. **Default:** *32*.

mkdocs.yml

@@ -54,6 +54,7 @@ nav:
   - Concepts:
       - ASGI: concepts/asgi.md
       - Lifespan: concepts/lifespan.md
+      - HTTP/2: concepts/http2.md
       - Logging: concepts/logging.md
       - WebSockets: concepts/websockets.md
       - Event Loop: concepts/event-loop.md

pyproject.toml

@@ -40,6 +40,7 @@ dependencies = [
 [project.optional-dependencies]
 standard = [
     "colorama>=0.4; sys_platform == 'win32'",
+    "h2>=4.2.0",
     "httptools>=0.6.3",
     "python-dotenv>=0.13",
     "PyYAML>=5.1",

tests/protocols/http_utils.py

@@ -0,0 +1,165 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from typing import Any
+
+
+class MockSSLObject:
+    def __init__(self, alpn_protocol: str | None = None):
+        self._alpn_protocol = alpn_protocol
+
+    def selected_alpn_protocol(self) -> str | None:
+        return self._alpn_protocol
+
+
+class MockTransport:
+    def __init__(
+        self,
+        sockname: tuple[str, int] | None = None,
+        peername: tuple[str, int] | None = None,
+        sslcontext: bool = False,
+        ssl_object: Any = None,
+    ):
+        self.sockname = ("127.0.0.1", 8000) if sockname is None else sockname
+        self.peername = ("127.0.0.1", 8001) if peername is None else peername
+        self.sslcontext = sslcontext
+        self._ssl_object = ssl_object
+        self.closed = False
+        self.buffer = b""
+        self.read_paused = False
+        self._protocol: asyncio.Protocol | None = None
+
+    def get_extra_info(self, key: Any):
+        return {
+            "sockname": self.sockname,
+            "peername": self.peername,
+            "sslcontext": self.sslcontext,
+            "ssl_object": self._ssl_object,
+        }.get(key)
+
+    def write(self, data: bytes):
+        assert not self.closed
+        self.buffer += data
+
+    def close(self):
+        assert not self.closed
+        self.closed = True
+
+    def pause_reading(self):
+        self.read_paused = True
+
+    def resume_reading(self):
+        self.read_paused = False
+
+    def is_closing(self):
+        return self.closed
+
+    def clear_buffer(self):
+        self.buffer = b""
+
+    def set_protocol(self, protocol: asyncio.Protocol):
+        self._protocol = protocol
+
+    def get_protocol(self) -> asyncio.Protocol | None:
+        return self._protocol
+
+
+class MockTimerHandle:
+    def __init__(
+        self,
+        loop_later_list: list[MockTimerHandle],
+        delay: float,
+        callback: Callable[[], None],
+        args: tuple[Any, ...],
+    ):
+        self.loop_later_list = loop_later_list
+        self.delay = delay
+        self.callback = callback
+        self.args = args
+        self.cancelled = False
+
+    def cancel(self):
+        if not self.cancelled:
+            self.cancelled = True
+            self.loop_later_list.remove(self)
+
+
+class MockTask:
+    def add_done_callback(self, callback: Callable[[], None]):
+        pass
+
+
+class MockLoop:
+    def __init__(self) -> None:
+        self._tasks: list[asyncio.Task[Any]] = []
+        self._later: list[MockTimerHandle] = []
+
+    def create_task(self, coroutine: Any, **kwargs: Any) -> Any:
+        self._tasks.insert(0, coroutine)
+        return MockTask()
+
+    def call_later(self, delay: float, callback: Callable[[], None], *args: Any) -> MockTimerHandle:
+        handle = MockTimerHandle(self._later, delay, callback, args)
+        self._later.insert(0, handle)
+        return handle
+
+    async def run_one(self) -> Any:
+        return await self._tasks.pop()
+
+    def run_later(self, with_delay: float) -> None:
+        later: list[MockTimerHandle] = []
+        for timer_handle in self._later:
+            if with_delay >= timer_handle.delay:
+                timer_handle.callback(*timer_handle.args)
+            else:
+                later.append(timer_handle)
+        self._later = later
+
+
+H2C_UPGRADE_REQUEST = b"\r\n".join(
+    [
+        b"GET / HTTP/1.1",
+        b"Host: example.org",
+        b"Connection: Upgrade, HTTP2-Settings",
+        b"Upgrade: h2c",
+        b"HTTP2-Settings: AAMAAABkAAQBAAAAAAIAAAAA",
+        b"",
+        b"",
+    ]
+)
+
+
+def h2c_upgrade_request(
+    *,
+    method: bytes = b"GET",
+    connection: bytes | None = b"Upgrade, HTTP2-Settings",
+    upgrade: bytes | None = b"h2c",
+    settings: bytes | None = b"AAMAAABkAAQBAAAAAAIAAAAA",
+    extra_settings: bytes | None = None,
+    content_length: bytes | None = None,
+    transfer_encoding: bytes | None = None,
+    body: bytes = b"",
+) -> bytes:
+    """Build an h2c upgrade request, optionally with malformed pieces.
+
+    Set any keyword to None to omit that header. `extra_settings` adds a second
+    HTTP2-Settings header (used to assert duplicate-header rejection).
+    `content_length` / `transfer_encoding` / `body` build a request that
+    carries a body so tests can assert the upgrade is refused.
+    """
+    lines = [method + b" / HTTP/1.1", b"Host: example.org"]
+    if connection is not None:
+        lines.append(b"Connection: " + connection)
+    if upgrade is not None:
+        lines.append(b"Upgrade: " + upgrade)
+    if settings is not None:
+        lines.append(b"HTTP2-Settings: " + settings)
+    if extra_settings is not None:
+        lines.append(b"HTTP2-Settings: " + extra_settings)
+    if content_length is not None:
+        lines.append(b"Content-Length: " + content_length)
+    if transfer_encoding is not None:
+        lines.append(b"Transfer-Encoding: " + transfer_encoding)
+    lines.extend([b"", body])
+    return b"\r\n".join(lines)

tests/protocols/test_http.py

@@ -5,11 +5,17 @@ import logging
 import socket
 import threading
 import time
-from collections.abc import Callable
 from typing import TYPE_CHECKING, Any, TypeAlias
 
 import pytest
 
+from tests.protocols.http_utils import (
+    H2C_UPGRADE_REQUEST,
+    MockLoop,
+    MockSSLObject,
+    MockTransport,
+    h2c_upgrade_request,
+)
 from tests.response import Response
 from uvicorn import Server
 from uvicorn._types import ASGIApplication, ASGIReceiveCallable, ASGISendCallable, Scope
@@ -26,7 +32,15 @@ try:
 except ModuleNotFoundError:  # pragma: no cover
     skip_if_no_httptools = pytest.mark.skipif(True, reason="httptools is not installed")
 
+try:
+    from uvicorn.protocols.http.h2_impl import H2Protocol
+
+    skip_if_no_h2 = pytest.mark.skipif(False, reason="h2 is installed")
+except ModuleNotFoundError:  # pragma: no cover
+    skip_if_no_h2 = pytest.mark.skipif(True, reason="h2 is not installed")
+
 if TYPE_CHECKING:
+    from uvicorn.protocols.http.h2_impl import H2Protocol
     from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
     from uvicorn.protocols.websockets.websockets_impl import WebSocketProtocol
     from uvicorn.protocols.websockets.wsproto_impl import WSProtocol as _WSProtocol
@@ -34,6 +48,7 @@ if TYPE_CHECKING:
     WSProtocol: TypeAlias = WebSocketProtocol | _WSProtocol
     HTTPProtocol: TypeAlias = H11Protocol | HttpToolsProtocol
 
+
 pytestmark = pytest.mark.anyio
 
 
@@ -122,18 +137,6 @@ UPGRADE_REQUEST = b"\r\n".join(
     ]
 )
 
-UPGRADE_HTTP2_REQUEST = b"\r\n".join(
-    [
-        b"GET / HTTP/1.1",
-        b"Host: example.org",
-        b"Connection: upgrade",
-        b"Upgrade: h2c",
-        b"Sec-WebSocket-Version: 11",
-        b"",
-        b"",
-    ]
-)
-
 INVALID_REQUEST_TEMPLATE = b"\r\n".join(
     [
         b"%s",
@@ -167,92 +170,6 @@ UPGRADE_REQUEST_ERROR_FIELD = b"\r\n".join(
 )
 
 
-class MockTransport:
-    def __init__(
-        self, sockname: tuple[str, int] | None = None, peername: tuple[str, int] | None = None, sslcontext: bool = False
-    ):
-        self.sockname = ("127.0.0.1", 8000) if sockname is None else sockname
-        self.peername = ("127.0.0.1", 8001) if peername is None else peername
-        self.sslcontext = sslcontext
-        self.closed = False
-        self.buffer = b""
-        self.read_paused = False
-
-    def get_extra_info(self, key: Any):
-        return {"sockname": self.sockname, "peername": self.peername, "sslcontext": self.sslcontext}.get(key)
-
-    def write(self, data: bytes):
-        assert not self.closed
-        self.buffer += data
-
-    def close(self):
-        assert not self.closed
-        self.closed = True
-
-    def pause_reading(self):
-        self.read_paused = True
-
-    def resume_reading(self):
-        self.read_paused = False
-
-    def is_closing(self):
-        return self.closed
-
-    def clear_buffer(self):
-        self.buffer = b""
-
-    def set_protocol(self, protocol: asyncio.Protocol):
-        pass
-
-
-class MockTimerHandle:
-    def __init__(
-        self, loop_later_list: list[MockTimerHandle], delay: float, callback: Callable[[], None], args: tuple[Any, ...]
-    ):
-        self.loop_later_list = loop_later_list
-        self.delay = delay
-        self.callback = callback
-        self.args = args
-        self.cancelled = False
-
-    def cancel(self):
-        if not self.cancelled:
-            self.cancelled = True
-            self.loop_later_list.remove(self)
-
-
-class MockLoop:
-    def __init__(self):
-        self._tasks: list[asyncio.Task[Any]] = []
-        self._later: list[MockTimerHandle] = []
-
-    def create_task(self, coroutine: Any, **kwargs: Any) -> Any:
-        self._tasks.insert(0, coroutine)
-        return MockTask()
-
-    def call_later(self, delay: float, callback: Callable[[], None], *args: Any) -> MockTimerHandle:
-        handle = MockTimerHandle(self._later, delay, callback, args)
-        self._later.insert(0, handle)
-        return handle
-
-    async def run_one(self):
-        return await self._tasks.pop()
-
-    def run_later(self, with_delay: float) -> None:
-        later: list[MockTimerHandle] = []
-        for timer_handle in self._later:
-            if with_delay >= timer_handle.delay:
-                timer_handle.callback(*timer_handle.args)
-            else:
-                later.append(timer_handle)
-        self._later = later
-
-
-class MockTask:
-    def add_done_callback(self, callback: Callable[[], None]):
-        pass
-
-
 class MockProtocol(asyncio.Protocol):
     loop: MockLoop
     transport: MockTransport
@@ -265,10 +182,15 @@ def get_connected_protocol(
     app: ASGIApplication,
     http_protocol_cls: type[HTTPProtocol],
     lifespan: LifespanOff | LifespanOn | None = None,
+    alpn_protocol: str | None = None,
     **kwargs: Any,
 ) -> MockProtocol:
     loop = MockLoop()
-    transport = MockTransport()
+    if alpn_protocol is not None:
+        ssl_object = MockSSLObject(alpn_protocol=alpn_protocol)
+        transport = MockTransport(sslcontext=True, ssl_object=ssl_object)
+    else:
+        transport = MockTransport()
     config = Config(app=app, **kwargs)
     lifespan = lifespan or LifespanOff(config)
     server_state = ServerState()
@@ -936,16 +858,6 @@ async def test_unsupported_ws_upgrade_request_warn_on_auto(
     assert msg in warnings
 
 
-async def test_http2_upgrade_request(http_protocol_cls: type[HTTPProtocol], ws_protocol_cls: type[WSProtocol]):
-    app = Response("Hello, world", media_type="text/plain")
-
-    protocol = get_connected_protocol(app, http_protocol_cls, ws=ws_protocol_cls)
-    protocol.data_received(UPGRADE_HTTP2_REQUEST)
-    await protocol.loop.run_one()
-    assert b"HTTP/1.1 200 OK" in protocol.transport.buffer
-    assert b"Hello, world" in protocol.transport.buffer
-
-
 async def asgi3app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable):
     pass
 
@@ -1210,6 +1122,129 @@ async def test_header_upgrade_is_not_websocket_depend_installed(
     assert b"Hello, world" in protocol.transport.buffer
 
 
+@skip_if_no_h2
+async def test_alpn_h2_upgrade(http_protocol_cls: type[HTTPProtocol]):
+    """Test that ALPN h2 negotiation switches to H2Protocol."""
+    app = Response("Hello, world", media_type="text/plain")
+    protocol = get_connected_protocol(app, http_protocol_cls, alpn_protocol="h2", http2=True)
+    assert isinstance(protocol.transport.get_protocol(), H2Protocol)
+
+
+@skip_if_no_h2
+async def test_alpn_h2_upgrade_not_triggered_without_h2_negotiation(http_protocol_cls: type[HTTPProtocol]):
+    """Test that ALPN upgrade doesn't happen when http/1.1 was negotiated."""
+    app = Response("Hello, world", media_type="text/plain")
+    protocol = get_connected_protocol(app, http_protocol_cls, alpn_protocol="http/1.1", http2=True)
+    assert protocol.transport.get_protocol() is None
+
+
+@skip_if_no_h2
+async def test_alpn_h2_upgrade_disabled_with_http2_false(http_protocol_cls: type[HTTPProtocol]):
+    """Test that ALPN h2 upgrade is disabled when http2=False."""
+    app = Response("Hello, world", media_type="text/plain")
+    protocol = get_connected_protocol(app, http_protocol_cls, alpn_protocol="h2", http2=False)
+    assert protocol.transport.get_protocol() is None
+
+
+@skip_if_no_h2
+async def test_h2c_upgrade(http_protocol_cls: type[HTTPProtocol]):
+    """Test HTTP/2 cleartext (h2c) upgrade via Upgrade header."""
+    app = Response("Hello, world", media_type="text/plain")
+    protocol = get_connected_protocol(app, http_protocol_cls, http2=True)
+    protocol.data_received(H2C_UPGRADE_REQUEST)
+
+    assert b"HTTP/1.1 101 Switching Protocols" in protocol.transport.buffer
+    assert b"Upgrade: h2c" in protocol.transport.buffer
+    assert isinstance(protocol.transport.get_protocol(), H2Protocol)
+
+    # Consume the H2 request task to avoid unawaited coroutine warning
+    h2_protocol = protocol.transport.get_protocol()
+    await h2_protocol.loop.run_one()  # type: ignore[union-attr]
+
+
+@skip_if_no_h2
+async def test_h2c_upgrade_disabled_with_http2_false(http_protocol_cls: type[HTTPProtocol]):
+    """Test that h2c upgrade is disabled when http2=False."""
+    app = Response("Hello, world", media_type="text/plain")
+    protocol = get_connected_protocol(app, http_protocol_cls, http2=False)
+    protocol.data_received(H2C_UPGRADE_REQUEST)
+    await protocol.loop.run_one()
+
+    assert b"HTTP/1.1 101 Switching Protocols" not in protocol.transport.buffer
+    assert b"HTTP/1.1 200 OK" in protocol.transport.buffer
+
+
+@skip_if_no_h2
+@pytest.mark.parametrize(
+    "request_bytes",
+    [
+        pytest.param(h2c_upgrade_request(settings=None), id="missing_http2_settings_header"),
+        pytest.param(h2c_upgrade_request(connection=b"Upgrade"), id="missing_connection_token"),
+        pytest.param(h2c_upgrade_request(settings=b""), id="empty_http2_settings_value"),
+        pytest.param(
+            h2c_upgrade_request(extra_settings=b"AAMAAABkAAQBAAAAAAIAAAAA"),
+            id="duplicate_http2_settings_header",
+        ),
+        # Requests with a body cannot be safely upgraded - the HTTP/1.1 body would
+        # need to feed stream 1 in HTTP/2, and we don't carry it across the switch.
+        pytest.param(
+            h2c_upgrade_request(method=b"POST", content_length=b"5", body=b"hello"),
+            id="content_length_body",
+        ),
+        pytest.param(
+            h2c_upgrade_request(method=b"POST", transfer_encoding=b"chunked"),
+            id="transfer_encoding_chunked",
+        ),
+    ],
+)
+async def test_h2c_upgrade_request_falls_back_to_http1(
+    http_protocol_cls: type[HTTPProtocol],
+    request_bytes: bytes,
+    caplog: pytest.LogCaptureFixture,
+):
+    """When the upgrade request is structurally malformed (missing required
+    headers, has a body), the server falls back to plain HTTP/1.1 instead of
+    sending `101 Switching Protocols`."""
+    caplog.set_level(logging.WARNING, logger="uvicorn.error")
+    app = Response("Hello, world", media_type="text/plain")
+    protocol = get_connected_protocol(app, http_protocol_cls, http2=True)
+    protocol.data_received(request_bytes)
+    await protocol.loop.run_one()
+
+    assert b"HTTP/1.1 101 Switching Protocols" not in protocol.transport.buffer
+    assert b"HTTP/1.1 200 OK" in protocol.transport.buffer
+    assert protocol.transport.get_protocol() is None
+    assert any("Ignoring h2c upgrade" in record.getMessage() for record in caplog.records)
+
+
+@skip_if_no_h2
+@pytest.mark.parametrize(
+    "request_bytes",
+    [
+        pytest.param(h2c_upgrade_request(settings=b"!!not-base64!!"), id="non_base64_http2_settings"),
+        # `AAAA` decodes to 3 bytes, which is shorter than one SETTINGS entry (6 bytes).
+        pytest.param(h2c_upgrade_request(settings=b"AAAA"), id="invalid_settings_frame_body"),
+        # `AAIAAAAC` encodes ENABLE_PUSH=2, which is a valid SETTINGS body shape but
+        # an invalid value (must be 0 or 1).
+        pytest.param(h2c_upgrade_request(settings=b"AAIAAAAC"), id="invalid_settings_value"),
+    ],
+)
+async def test_h2c_upgrade_with_invalid_settings_returns_400(
+    http_protocol_cls: type[HTTPProtocol],
+    request_bytes: bytes,
+):
+    """When the HTTP2-Settings payload is well-framed enough to attempt the
+    upgrade but hyper-h2 rejects the SETTINGS contents, the server replies
+    with 400 instead of leaving the wire in a half-broken state."""
+    app = Response("Hello, world", media_type="text/plain")
+    protocol = get_connected_protocol(app, http_protocol_cls, http2=True)
+    protocol.data_received(request_bytes)
+
+    assert b"HTTP/1.1 400" in protocol.transport.buffer
+    assert b"HTTP/1.1 101 Switching Protocols" not in protocol.transport.buffer
+    assert protocol.transport.get_protocol() is None
+
+
 async def test_header_upgrade_is_websocket_depend_not_installed(
     caplog: pytest.LogCaptureFixture, http_protocol_cls: type[HTTPProtocol]
 ):

tests/test_config.py

@@ -6,6 +6,7 @@ import json
 import logging
 import os
 import socket
+import ssl
 import sys
 from collections.abc import Callable, Iterator
 from contextlib import closing
@@ -620,3 +621,58 @@ def test_setup_event_loop_is_removed(caplog: pytest.LogCaptureFixture) -> None:
         AttributeError, match="The `setup_event_loop` method was replaced by `get_loop_factory` in uvicorn 0.36.0."
     ):
         config.setup_event_loop()
+
+
+def test_http2_with_ssl_sets_alpn(
+    tls_ca_certificate_pem_path: str,
+    tls_ca_certificate_private_key_path: str,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test that http2=True with SSL configures ALPN protocols."""
+    recorded: list[list[str]] = []
+    original = ssl.SSLContext.set_alpn_protocols
+
+    def spy(self: ssl.SSLContext, protocols: list[str]) -> None:
+        recorded.append(protocols)
+        original(self, protocols)
+
+    monkeypatch.setattr(ssl.SSLContext, "set_alpn_protocols", spy)
+
+    config = Config(
+        app=asgi_app,
+        http2=True,
+        ssl_certfile=tls_ca_certificate_pem_path,
+        ssl_keyfile=tls_ca_certificate_private_key_path,
+    )
+    config.load()
+
+    assert config.is_ssl is True
+    assert config.ssl is not None
+    assert config.h2_protocol_class is not None
+    assert ["h2", "http/1.1"] in recorded
+
+
+def test_http2_as_string_path() -> None:
+    """Test that http2 can be specified as a string import path."""
+    config = Config(
+        app=asgi_app,
+        http2="uvicorn.protocols.http.h2_impl:H2Protocol",
+    )
+    config.load()
+
+    from uvicorn.protocols.http.h2_impl import H2Protocol
+
+    assert config.h2_protocol_class is H2Protocol
+
+
+def test_http2_as_class() -> None:
+    """Test that http2 can be specified as a protocol class directly."""
+    from uvicorn.protocols.http.h2_impl import H2Protocol
+
+    config = Config(
+        app=asgi_app,
+        http2=H2Protocol,
+    )
+    config.load()
+
+    assert config.h2_protocol_class is H2Protocol

uv.lock

@@ -574,6 +574,28 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" },
 ]
 
+[[package]]
+name = "h2"
+version = "4.3.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "hpack" },
+    { name = "hyperframe" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/1d/17/afa56379f94ad0fe8defd37d6eb3f89a25404ffc71d4d848893d270325fc/h2-4.3.0.tar.gz", hash = "sha256:6c59efe4323fa18b47a632221a1888bd7fde6249819beda254aeca909f221bf1", size = 2152026, upload-time = "2025-08-23T18:12:19.778Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/69/b2/119f6e6dcbd96f9069ce9a2665e0146588dc9f88f29549711853645e736a/h2-4.3.0-py3-none-any.whl", hash = "sha256:c438f029a25f7945c69e0ccf0fb951dc3f73a5f6412981daee861431b70e2bdd", size = 61779, upload-time = "2025-08-23T18:12:17.779Z" },
+]
+
+[[package]]
+name = "hpack"
+version = "4.1.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/2c/48/71de9ed269fdae9c8057e5a4c0aa7402e8bb16f2c6e90b3aa53327b113f8/hpack-4.1.0.tar.gz", hash = "sha256:ec5eca154f7056aa06f196a557655c5b009b382873ac8d1e66e79e87535f1dca", size = 51276, upload-time = "2025-01-22T21:44:58.347Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/07/c6/80c95b1b2b94682a72cbdbfb85b81ae2daffa4291fbfa1b1464502ede10d/hpack-4.1.0-py3-none-any.whl", hash = "sha256:157ac792668d995c657d93111f46b4535ed114f0c9c8d672271bbec7eae1b496", size = 34357, upload-time = "2025-01-22T21:44:56.92Z" },
+]
+
 [[package]]
 name = "httpcore"
 version = "1.0.9"
@@ -645,6 +667,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" },
 ]
 
+[[package]]
+name = "hyperframe"
+version = "6.1.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/02/e7/94f8232d4a74cc99514c13a9f995811485a6903d48e5d952771ef6322e30/hyperframe-6.1.0.tar.gz", hash = "sha256:f630908a00854a7adeabd6382b43923a4c4cd4b821fcb527e6ab9e15382a3b08", size = 26566, upload-time = "2025-01-22T21:41:49.302Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/48/30/47d0bf6072f7252e6521f3447ccfa40b421b6824517f82854703d0f5a98b/hyperframe-6.1.0-py3-none-any.whl", hash = "sha256:b03380493a519fce58ea5af42e4a42317bf9bd425596f7a0835ffce80f1a42e5", size = 13007, upload-time = "2025-01-22T21:41:47.295Z" },
+]
+
 [[package]]
 name = "id"
 version = "1.6.1"
@@ -1776,6 +1807,7 @@ dependencies = [
 [package.optional-dependencies]
 standard = [
     { name = "colorama", marker = "sys_platform == 'win32'" },
+    { name = "h2" },
     { name = "httptools" },
     { name = "python-dotenv" },
     { name = "pyyaml" },
@@ -1818,6 +1850,7 @@ requires-dist = [
     { name = "click", specifier = ">=7.0" },
     { name = "colorama", marker = "sys_platform == 'win32' and extra == 'standard'", specifier = ">=0.4" },
     { name = "h11", specifier = ">=0.8" },
+    { name = "h2", marker = "extra == 'standard'", specifier = ">=4.2.0" },
     { name = "httptools", marker = "extra == 'standard'", specifier = ">=0.6.3" },
     { name = "python-dotenv", marker = "extra == 'standard'", specifier = ">=0.13" },
     { name = "pyyaml", marker = "extra == 'standard'", specifier = ">=5.1" },

uvicorn/config.py

@@ -12,7 +12,7 @@ import sys
 from collections.abc import Awaitable, Callable
 from configparser import RawConfigParser
 from pathlib import Path
-from typing import IO, Any, Literal
+from typing import IO, TYPE_CHECKING, Any, Literal
 
 import click
 
@@ -25,6 +25,9 @@ from uvicorn.middleware.message_logger import MessageLoggerMiddleware
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 from uvicorn.middleware.wsgi import WSGIMiddleware
 
+if TYPE_CHECKING:
+    from uvicorn.protocols.http.h2_impl import HTTP2Protocol
+
 HTTPProtocolType = Literal["auto", "h11", "httptools"]
 WSProtocolType = Literal["auto", "none", "websockets", "websockets-sansio", "wsproto"]
 LifespanType = Literal["auto", "on", "off"]
@@ -110,6 +113,7 @@ def create_ssl_context(
     cert_reqs: int,
     ca_certs: str | os.PathLike[str] | None,
     ciphers: str | None,
+    alpn_protocols: list[str] | None = None,
 ) -> ssl.SSLContext:
     ctx = ssl.SSLContext(ssl_version)
     get_password = (lambda: password) if password else None
@@ -119,6 +123,8 @@ def create_ssl_context(
         ctx.load_verify_locations(ca_certs)
     if ciphers:
         ctx.set_ciphers(ciphers)
+    if alpn_protocols:
+        ctx.set_alpn_protocols(alpn_protocols)
     return ctx
 
 
@@ -154,7 +160,7 @@ def resolve_reload_patterns(patterns_list: list[str], directories_list: list[str
     directories = list(map(lambda x: x.resolve(), directories))
     directories = list({reload_path for reload_path in directories if is_dir(reload_path)})
 
-    children = []
+    children: list[Path] = []
     for j in range(len(directories)):
         for k in range(j + 1, len(directories)):  # pragma: full coverage
             if directories[j] in directories[k].parents:
@@ -185,6 +191,7 @@ class Config:
         fd: int | None = None,
         loop: LoopFactoryType | str = "auto",
         http: type[asyncio.Protocol] | HTTPProtocolType | str = "auto",
+        http2: bool | type[HTTP2Protocol] | str = False,
         ws: type[asyncio.Protocol] | WSProtocolType | str = "auto",
         ws_max_size: int = 16 * 1024 * 1024,
         ws_max_queue: int = 32,
@@ -236,6 +243,7 @@ class Config:
         self.fd = fd
         self.loop = loop
         self.http = http
+        self.http2 = http2
         self.ws = ws
         self.ws_max_size = ws_max_size
         self.ws_max_queue = ws_max_queue
@@ -407,6 +415,9 @@ class Config:
 
         if self.is_ssl:
             assert self.ssl_certfile
+            alpn_protocols: list[str] | None = None
+            if self.http2:
+                alpn_protocols = ["h2", "http/1.1"]
             self.ssl: ssl.SSLContext | None = create_ssl_context(
                 keyfile=self.ssl_keyfile,
                 certfile=self.ssl_certfile,
@@ -415,6 +426,7 @@ class Config:
                 cert_reqs=self.ssl_cert_reqs,
                 ca_certs=self.ssl_ca_certs,
                 ciphers=self.ssl_ciphers,
+                alpn_protocols=alpn_protocols,
             )
         else:
             self.ssl = None
@@ -432,6 +444,15 @@ class Config:
         else:
             self.http_protocol_class = self.http
 
+        if self.http2 is False:
+            self.h2_protocol_class: type[HTTP2Protocol] | None = None
+        elif self.http2 is True:
+            self.h2_protocol_class = import_from_string("uvicorn.protocols.http.h2_impl:H2Protocol")
+        elif isinstance(self.http2, str):
+            self.h2_protocol_class = import_from_string(self.http2)
+        else:
+            self.h2_protocol_class = self.http2
+
         if isinstance(self.ws, str):
             ws_protocol_class = import_from_string(WS_PROTOCOLS.get(self.ws, self.ws))
             self.ws_protocol_class: type[asyncio.Protocol] | None = ws_protocol_class

uvicorn/main.py

@@ -9,7 +9,7 @@ import sys
 import warnings
 from collections.abc import Callable
 from configparser import RawConfigParser
-from typing import IO, Any, get_args
+from typing import IO, TYPE_CHECKING, Any, get_args
 
 import click
 
@@ -31,6 +31,9 @@ from uvicorn.config import (
 from uvicorn.server import Server
 from uvicorn.supervisors import ChangeReload, Multiprocess
 
+if TYPE_CHECKING:
+    from uvicorn.protocols.http.h2_impl import HTTP2Protocol
+
 LEVEL_CHOICES = click.Choice(list(LOG_LEVELS.keys()))
 LIFESPAN_CHOICES = click.Choice(list(LIFESPAN.keys()))
 INTERFACE_CHOICES = click.Choice(INTERFACES)
@@ -132,6 +135,12 @@ def print_version(ctx: click.Context, param: click.Parameter, value: bool) -> No
     help="HTTP protocol implementation.",
     show_default=True,
 )
+@click.option(
+    "--http2",
+    is_flag=True,
+    default=False,
+    help="Enable HTTP/2 support.",
+)
 @click.option(
     "--ws",
     type=str,
@@ -387,6 +396,7 @@ def main(
     fd: int,
     loop: LoopFactoryType | str,
     http: HTTPProtocolType | str,
+    http2: bool,
     ws: WSProtocolType | str,
     ws_max_size: int,
     ws_max_queue: int,
@@ -438,6 +448,7 @@ def main(
         fd=fd,
         loop=loop,
         http=http,
+        http2=http2,
         ws=ws,
         ws_max_size=ws_max_size,
         ws_max_queue=ws_max_queue,
@@ -492,6 +503,7 @@ def run(
     fd: int | None = None,
     loop: LoopFactoryType | str = "auto",
     http: type[asyncio.Protocol] | HTTPProtocolType | str = "auto",
+    http2: bool | type[HTTP2Protocol] | str = False,
     ws: type[asyncio.Protocol] | WSProtocolType | str = "auto",
     ws_max_size: int = 16777216,
     ws_max_queue: int = 32,
@@ -546,6 +558,7 @@ def run(
         fd=fd,
         loop=loop,
         http=http,
+        http2=http2,
         ws=ws,
         ws_max_size=ws_max_size,
         ws_max_queue=ws_max_queue,

uvicorn/protocols/http/h11_impl.py

@@ -6,6 +6,7 @@ import http
 import logging
 import sys
 from collections.abc import Callable
+from ssl import SSLObject
 from typing import Any, Literal
 from urllib.parse import unquote
 
@@ -62,6 +63,7 @@ class H11Protocol(asyncio.Protocol):
             else DEFAULT_MAX_INCOMPLETE_EVENT_SIZE,
         )
         self.ws_protocol_class = config.ws_protocol_class
+        self.h2_protocol_class = config.h2_protocol_class
         self.root_path = config.root_path
         self.limit_concurrency = config.limit_concurrency
         self.app_state = app_state
@@ -86,6 +88,8 @@ class H11Protocol(asyncio.Protocol):
         self.scope: HTTPScope = None  # type: ignore[assignment]
         self.headers: list[tuple[bytes, bytes]] = None  # type: ignore[assignment]
         self.cycle: RequestResponseCycle = None  # type: ignore[assignment]
+        # Cached HTTP2-Settings header value when an h2c upgrade has been validated.
+        self._h2c_settings: bytes | None = None
 
     # Protocol interface
     def connection_made(  # type: ignore[override]
@@ -99,10 +103,37 @@ class H11Protocol(asyncio.Protocol):
         self.client = get_remote_addr(transport)
         self.scheme = "https" if is_ssl(transport) else "http"
 
+        # Check for ALPN negotiation - if h2 was negotiated, switch to HTTP/2 protocol
+        if self._should_upgrade_to_h2(transport):
+            self._upgrade_to_h2(transport)
+            return
+
         if self.logger.level <= TRACE_LOG_LEVEL:
             prefix = "%s:%d - " % self.client if self.client else ""
             self.logger.log(TRACE_LOG_LEVEL, "%sHTTP connection made", prefix)
 
+    def _should_upgrade_to_h2(self, transport: asyncio.Transport) -> bool:
+        """Check if the connection should be upgraded to HTTP/2 based on ALPN."""
+        if self.h2_protocol_class is None:
+            return False
+        ssl_object: SSLObject | None = transport.get_extra_info("ssl_object")
+        selected_protocol = ssl_object and ssl_object.selected_alpn_protocol()
+        return selected_protocol == "h2"
+
+    def _upgrade_to_h2(self, transport: asyncio.Transport) -> None:
+        """Upgrade the connection to HTTP/2 protocol."""
+        assert self.h2_protocol_class is not None
+        self.connections.discard(self)
+
+        h2_protocol = self.h2_protocol_class(
+            config=self.config,
+            server_state=self.server_state,
+            app_state=self.app_state,
+            _loop=self.loop,
+        )
+        transport.set_protocol(h2_protocol)
+        h2_protocol.connection_made(transport)
+
     def connection_lost(self, exc: Exception | None) -> None:
         self.connections.discard(self)
 
@@ -136,23 +167,34 @@ class H11Protocol(asyncio.Protocol):
             self.timeout_keep_alive_task.cancel()
             self.timeout_keep_alive_task = None
 
-    def _get_upgrade(self) -> bytes | None:
-        connection = []
+    def _get_upgrade(self) -> tuple[bytes | None, list[bytes]]:
+        connection: list[bytes] = []
         upgrade = None
         for name, value in self.headers:
             if name == b"connection":
-                connection = [token.lower().strip() for token in value.split(b",")]
+                connection.extend(token.lower().strip() for token in value.split(b","))
             if name == b"upgrade":
                 upgrade = value.lower()
         if b"upgrade" in connection:
-            return upgrade
-        return None
+            return upgrade, connection
+        return None, connection
 
     def _should_upgrade_to_ws(self) -> bool:
         if self.ws_protocol_class is None:
             return False
         return True
 
+    def _should_upgrade_to_h2c(self) -> bool:
+        """Check if h2 support is enabled for h2c upgrade.
+
+        h2c is HTTP/2 cleartext only; refuse it on TLS connections so a client
+        cannot bypass ALPN by sending `Upgrade: h2c` over a session that
+        negotiated `http/1.1`. HTTP/2 over TLS must come through ALPN.
+        """
+        if self.h2_protocol_class is None:
+            return False
+        return not is_ssl(self.transport)
+
     def _unsupported_upgrade_warning(self) -> None:
         msg = "Unsupported upgrade request."
         self.logger.warning(msg)
@@ -160,13 +202,42 @@ class H11Protocol(asyncio.Protocol):
             msg = "No supported WebSocket library detected. Please use \"pip install 'uvicorn[standard]'\", or install 'websockets' or 'wsproto' manually."  # noqa: E501
             self.logger.warning(msg)
 
-    def _should_upgrade(self) -> bool:
-        upgrade = self._get_upgrade()
+    def _get_h2c_settings(self, connection_tokens: list[bytes]) -> bytes | None:
+        # Per RFC 7540 section 3.2, an h2c upgrade requires the `HTTP2-Settings`
+        # connection-option and exactly one `HTTP2-Settings` header field. Requests
+        # that carry a body cannot be upgraded because we'd lose the body bytes
+        # when we hand stream 1 to h2. The actual SETTINGS payload is validated
+        # later by `H2Protocol.initiate_h2c_upgrade`, which only commits the
+        # protocol switch if hyper-h2 accepts the payload.
+        if b"http2-settings" not in connection_tokens:
+            return None
+        seen: bytes | None = None
+        for name, value in self.headers:
+            if name == b"http2-settings":
+                if seen is not None or not value:
+                    return None
+                seen = value
+            elif name == b"content-length" and value not in (b"", b"0"):
+                return None
+            elif name == b"transfer-encoding":
+                return None
+        return seen
+
+    def _get_upgrade_type(self) -> str | None:
+        """Determine the type of upgrade request: 'websocket', 'h2c', or None."""
+        upgrade, connection_tokens = self._get_upgrade()
         if upgrade == b"websocket" and self._should_upgrade_to_ws():
-            return True
+            return "websocket"
+        if upgrade == b"h2c" and self._should_upgrade_to_h2c():
+            settings = self._get_h2c_settings(connection_tokens)
+            if settings is not None:
+                self._h2c_settings = settings
+                return "h2c"
+            self.logger.warning("Ignoring h2c upgrade with missing or invalid HTTP2-Settings header.")
+            return None
         if upgrade is not None:
             self._unsupported_upgrade_warning()
-        return False
+        return None
 
     def data_received(self, data: bytes) -> None:
         self._unset_keepalive_if_required()
@@ -216,9 +287,13 @@ class H11Protocol(asyncio.Protocol):
                     "headers": self.headers,
                     "state": self.app_state.copy(),
                 }
-                if self._should_upgrade():
+                upgrade_type = self._get_upgrade_type()
+                if upgrade_type == "websocket":
                     self.handle_websocket_upgrade(event)
                     return
+                elif upgrade_type == "h2c":
+                    self.handle_h2c_upgrade(event)
+                    return
 
                 # Handle 503 responses when 'limit_concurrency' is exceeded.
                 if self.limit_concurrency is not None and (
@@ -297,6 +372,47 @@ class H11Protocol(asyncio.Protocol):
         protocol.data_received(b"".join(output))
         self.transport.set_protocol(protocol)
 
+    def handle_h2c_upgrade(self, event: h11.Request) -> None:
+        """Handle HTTP/2 cleartext (h2c) upgrade request."""
+        assert self.h2_protocol_class is not None
+        assert self._h2c_settings is not None
+        if self.logger.level <= TRACE_LOG_LEVEL:  # pragma: full coverage
+            prefix = "%s:%d - " % self.client if self.client else ""
+            self.logger.log(TRACE_LOG_LEVEL, "%sUpgrading to HTTP/2 (h2c)", prefix)
+
+        # Bytes the peer already sent after the upgrade request - typically
+        # the HTTP/2 client preface (`PRI * HTTP/2.0...`) and the first
+        # frames - need to be forwarded to the new protocol after the switch
+        # so the connection doesn't stall on lost initial frames.
+        trailing = self.conn.trailing_data[0]
+
+        h2_protocol = self.h2_protocol_class(
+            config=self.config,
+            server_state=self.server_state,
+            app_state=self.app_state,
+            _loop=self.loop,
+        )
+
+        # Try the upgrade first; only commit `101 Switching Protocols` if h2
+        # accepts the SETTINGS payload. Otherwise the wire would be left in
+        # a half-broken state with the peer expecting HTTP/2 and the server
+        # unable to speak it.
+        if not h2_protocol.initiate_h2c_upgrade(
+            self.transport,
+            event.method.decode("ascii"),
+            event.target.decode("ascii"),
+            self.headers,
+            self._h2c_settings,
+        ):
+            self.send_400_response("Invalid HTTP2-Settings header for h2c upgrade")
+            return
+
+        self.connections.discard(self)
+        self.transport.write(b"HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: h2c\r\n\r\n")
+        self.transport.set_protocol(h2_protocol)
+        if trailing:
+            h2_protocol.data_received(trailing)
+
     def send_400_response(self, msg: str) -> None:
         reason = STATUS_PHRASES[400]
         headers: list[tuple[bytes, bytes]] = [

uvicorn/protocols/http/httptools_impl.py

@@ -10,6 +10,7 @@ import urllib
 from asyncio.events import TimerHandle
 from collections import deque
 from collections.abc import Callable
+from ssl import SSLObject
 from typing import Any, Literal
 
 import httptools
@@ -69,6 +70,7 @@ class HttpToolsProtocol(asyncio.Protocol):
             pass
 
         self.ws_protocol_class = config.ws_protocol_class
+        self.h2_protocol_class = config.h2_protocol_class
         self.root_path = config.root_path
         self.limit_concurrency = config.limit_concurrency
         self.app_state = app_state
@@ -95,6 +97,8 @@ class HttpToolsProtocol(asyncio.Protocol):
         self.headers: list[tuple[bytes, bytes]] = None  # type: ignore[assignment]
         self.expect_100_continue = False
         self.cycle: RequestResponseCycle = None  # type: ignore[assignment]
+        # Cached HTTP2-Settings header value when an h2c upgrade has been validated.
+        self._h2c_settings: bytes | None = None
 
     # Protocol interface
     def connection_made(  # type: ignore[override]
@@ -108,10 +112,37 @@ class HttpToolsProtocol(asyncio.Protocol):
         self.client = get_remote_addr(transport)
         self.scheme = "https" if is_ssl(transport) else "http"
 
+        # Check for ALPN negotiation - if h2 was negotiated, switch to HTTP/2 protocol
+        if self._should_upgrade_to_h2(transport):
+            self._upgrade_to_h2(transport)
+            return
+
         if self.logger.level <= TRACE_LOG_LEVEL:
             prefix = "%s:%d - " % self.client if self.client else ""
             self.logger.log(TRACE_LOG_LEVEL, "%sHTTP connection made", prefix)
 
+    def _should_upgrade_to_h2(self, transport: asyncio.Transport) -> bool:
+        """Check if the connection should be upgraded to HTTP/2 based on ALPN."""
+        if self.h2_protocol_class is None:
+            return False
+        ssl_object: SSLObject | None = transport.get_extra_info("ssl_object")
+        selected_protocol = ssl_object and ssl_object.selected_alpn_protocol()
+        return selected_protocol == "h2"
+
+    def _upgrade_to_h2(self, transport: asyncio.Transport) -> None:
+        """Upgrade the connection to HTTP/2 protocol."""
+        assert self.h2_protocol_class is not None
+        self.connections.discard(self)
+
+        h2_protocol = self.h2_protocol_class(
+            config=self.config,
+            server_state=self.server_state,
+            app_state=self.app_state,
+            _loop=self.loop,
+        )
+        transport.set_protocol(h2_protocol)
+        h2_protocol.connection_made(transport)
+
     def connection_lost(self, exc: Exception | None) -> None:
         self.connections.discard(self)
 
@@ -139,32 +170,79 @@ class HttpToolsProtocol(asyncio.Protocol):
             self.timeout_keep_alive_task.cancel()
             self.timeout_keep_alive_task = None
 
-    def _get_upgrade(self) -> bytes | None:
-        connection = []
+    def _get_upgrade(self) -> tuple[bytes | None, list[bytes]]:
+        connection: list[bytes] = []
         upgrade = None
         for name, value in self.headers:
             if name == b"connection":
-                connection = [token.lower().strip() for token in value.split(b",")]
+                connection.extend(token.lower().strip() for token in value.split(b","))
             if name == b"upgrade":
                 upgrade = value.lower()
         if b"upgrade" in connection:
-            return upgrade
-        return None  # pragma: full coverage
+            return upgrade, connection
+        return None, connection  # pragma: full coverage
 
     def _should_upgrade_to_ws(self) -> bool:
         if self.ws_protocol_class is None:
             return False
         return True
 
+    def _should_upgrade_to_h2c(self) -> bool:
+        """Check if HTTP/2 protocol is available for h2c upgrade.
+
+        h2c is HTTP/2 cleartext only; refuse it on TLS connections so a client
+        cannot bypass ALPN by sending `Upgrade: h2c` over a session that
+        negotiated `http/1.1`. HTTP/2 over TLS must come through ALPN.
+        """
+        if self.h2_protocol_class is None:
+            return False
+        return not is_ssl(self.transport)
+
     def _unsupported_upgrade_warning(self) -> None:
         self.logger.warning("Unsupported upgrade request.")
         if not self._should_upgrade_to_ws():
             msg = "No supported WebSocket library detected. Please use \"pip install 'uvicorn[standard]'\", or install 'websockets' or 'wsproto' manually."  # noqa: E501
             self.logger.warning(msg)
 
+    def _get_h2c_settings(self, connection_tokens: list[bytes]) -> bytes | None:
+        # Per RFC 7540 section 3.2, an h2c upgrade requires the `HTTP2-Settings`
+        # connection-option and exactly one `HTTP2-Settings` header field. Requests
+        # that carry a body cannot be upgraded because we'd lose the body bytes
+        # when we hand stream 1 to h2. The actual SETTINGS payload is validated
+        # later by `H2Protocol.initiate_h2c_upgrade`, which only commits the
+        # protocol switch if hyper-h2 accepts the payload.
+        if b"http2-settings" not in connection_tokens:
+            return None
+        seen: bytes | None = None
+        for name, value in self.headers:
+            if name == b"http2-settings":
+                if seen is not None or not value:
+                    return None
+                seen = value
+            elif name == b"content-length" and value not in (b"", b"0"):
+                return None
+            elif name == b"transfer-encoding":
+                return None
+        return seen
+
+    def _get_upgrade_type(self) -> str | None:
+        """Determine the type of upgrade request: 'websocket', 'h2c', or None."""
+        upgrade, connection_tokens = self._get_upgrade()
+        if upgrade == b"websocket" and self._should_upgrade_to_ws():
+            return "websocket"
+        if upgrade == b"h2c" and self._should_upgrade_to_h2c():
+            settings = self._get_h2c_settings(connection_tokens)
+            if settings is not None:
+                self._h2c_settings = settings
+                return "h2c"
+            self.logger.warning("Ignoring h2c upgrade with missing or invalid HTTP2-Settings header.")
+            return None
+        if upgrade is not None:
+            self._unsupported_upgrade_warning()
+        return None
+
     def _should_upgrade(self) -> bool:
-        upgrade = self._get_upgrade()
-        return upgrade == b"websocket" and self._should_upgrade_to_ws()
+        return self._get_upgrade_type() is not None
 
     def data_received(self, data: bytes) -> None:
         self._unset_keepalive_if_required()
@@ -176,11 +254,18 @@ class HttpToolsProtocol(asyncio.Protocol):
             self.logger.warning(msg)
             self.send_400_response(msg)
             return
-        except httptools.HttpParserUpgrade:
-            if self._should_upgrade():
+        except httptools.HttpParserUpgrade as exc:
+            upgrade_type = self._get_upgrade_type()
+            # Anything past the offset reported by httptools is data the peer
+            # has already shipped after the upgrade headers (e.g. the HTTP/2
+            # client preface arriving in the same TCP packet as the Upgrade
+            # request). Hand it to the upgraded protocol once we switch.
+            offset = exc.args[0] if exc.args else len(data)
+            trailing = data[offset:]
+            if upgrade_type == "websocket":
                 self.handle_websocket_upgrade()
-            else:
-                self._unsupported_upgrade_warning()
+            elif upgrade_type == "h2c":
+                self.handle_h2c_upgrade(trailing)
 
     def handle_websocket_upgrade(self) -> None:
         if self.logger.level <= TRACE_LOG_LEVEL:
@@ -202,6 +287,47 @@ class HttpToolsProtocol(asyncio.Protocol):
         protocol.data_received(b"".join(output))
         self.transport.set_protocol(protocol)
 
+    def handle_h2c_upgrade(self, trailing_data: bytes = b"") -> None:
+        """Handle HTTP/2 cleartext (h2c) upgrade request.
+
+        `trailing_data` carries any bytes the peer sent after the upgrade
+        request in the same TCP read; they are forwarded to the upgraded
+        protocol's `data_received` after the switch so HTTP/2 frames sent
+        immediately after the Upgrade headers are not lost.
+        """
+        assert self.h2_protocol_class is not None
+        assert self._h2c_settings is not None
+        if self.logger.level <= TRACE_LOG_LEVEL:  # pragma: full coverage
+            prefix = "%s:%d - " % self.client if self.client else ""
+            self.logger.log(TRACE_LOG_LEVEL, "%sUpgrading to HTTP/2 (h2c)", prefix)
+
+        h2_protocol = self.h2_protocol_class(
+            config=self.config,
+            server_state=self.server_state,
+            app_state=self.app_state,
+            _loop=self.loop,
+        )
+
+        # Try the upgrade first; only commit `101 Switching Protocols` if h2
+        # accepts the SETTINGS payload. Otherwise the wire would be left in
+        # a half-broken state with the peer expecting HTTP/2 and the server
+        # unable to speak it.
+        if not h2_protocol.initiate_h2c_upgrade(
+            self.transport,
+            self.scope["method"],
+            self.url.decode("ascii"),
+            list(self.scope["headers"]),
+            self._h2c_settings,
+        ):
+            self.send_400_response("Invalid HTTP2-Settings header for h2c upgrade")
+            return
+
+        self.connections.discard(self)
+        self.transport.write(b"HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: h2c\r\n\r\n")
+        self.transport.set_protocol(h2_protocol)
+        if trailing_data:
+            h2_protocol.data_received(trailing_data)
+
     def send_400_response(self, msg: str) -> None:
         content = [STATUS_LINE[400]]
         for name, value in self.server_state.default_headers:

uvicorn/server.py

@@ -23,13 +23,16 @@ from uvicorn._compat import asyncio_run
 from uvicorn.config import Config
 
 if TYPE_CHECKING:
+    from uvicorn.protocols.http.h2_impl import H2Protocol
     from uvicorn.protocols.http.h11_impl import H11Protocol
     from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
     from uvicorn.protocols.websockets.websockets_impl import WebSocketProtocol
     from uvicorn.protocols.websockets.websockets_sansio_impl import WebSocketsSansIOProtocol
     from uvicorn.protocols.websockets.wsproto_impl import WSProtocol
 
-    Protocols: TypeAlias = H11Protocol | HttpToolsProtocol | WSProtocol | WebSocketProtocol | WebSocketsSansIOProtocol
+    Protocols: TypeAlias = (
+        H11Protocol | H2Protocol | HttpToolsProtocol | WSProtocol | WebSocketProtocol | WebSocketsSansIOProtocol
+    )
 
 HANDLED_SIGNALS = (
     signal.SIGINT,  # Unix signal 2. Sent by Ctrl+C.