0
# Middleware Components
1

2
ASGI middleware components for proxy header handling, protocol adapters, and debugging support.
3

4
## Imports
5

6
```python
7
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
8
from uvicorn.middleware.asgi2 import ASGI2Middleware
9
from uvicorn.middleware.wsgi import WSGIMiddleware
10
from uvicorn.middleware.message_logger import MessageLoggerMiddleware
11
```
12

13
## Capabilities
14

15
### Proxy Headers Middleware
16

17
Middleware for handling X-Forwarded-Proto and X-Forwarded-For proxy headers.
18

19
```python { .api }
20
class ProxyHeadersMiddleware:
21
    """
22
    Middleware for handling proxy headers.
23

24
    Parses X-Forwarded-Proto and X-Forwarded-For headers from trusted
25
    proxies and updates the ASGI scope with the client's real IP address
26
    and protocol scheme.
27

28
    This is essential when running behind reverse proxies like Nginx,
29
    Apache, or cloud load balancers.
30
    """
31

32
    def __init__(
33
        self,
34
        app: ASGI3Application,
35
        trusted_hosts: list[str] | str = "127.0.0.1",
36
    ) -> None:
37
        """
38
        Initialize proxy headers middleware.
39

40
        Args:
41
            app: ASGI application to wrap
42
            trusted_hosts: Trusted proxy hosts/networks as comma-separated string or list
43
                          Supports:
44
                          - "*" to trust all proxies (not recommended in production)
45
                          - IP addresses: "127.0.0.1", "192.168.1.1"
46
                          - CIDR networks: "10.0.0.0/8", "172.16.0.0/12"
47
                          - Multiple values: "127.0.0.1,10.0.0.0/8"
48
                          Default: "127.0.0.1"
49

50
        The middleware only processes headers from trusted proxy addresses.
51
        Headers from untrusted sources are ignored for security.
52
        """
53

54
    async def __call__(
55
        self,
56
        scope: Scope,
57
        receive: ASGIReceiveCallable,
58
        send: ASGISendCallable,
59
    ) -> None:
60
        """
61
        Process ASGI connection with proxy header handling.
62

63
        Args:
64
            scope: ASGI connection scope
65
            receive: Receive callable for incoming messages
66
            send: Send callable for outgoing messages
67

68
        For HTTP and WebSocket scopes, this middleware:
69
        1. Checks if the connection is from a trusted proxy
70
        2. Parses X-Forwarded-Proto header to update scope["scheme"]
71
        3. Parses X-Forwarded-For header to update scope["client"]
72
        4. Calls the wrapped application with updated scope
73
        """
74
```
75

76
### ASGI2 Middleware
77

78
Adapter to run ASGI2 applications as ASGI3.
79

80
```python { .api }
81
class ASGI2Middleware:
82
    """
83
    Adapter to run ASGI2 applications as ASGI3.
84

85
    ASGI2 uses a class-based interface where the application is instantiated
86
    with the scope and then called with receive/send. ASGI3 uses a single
87
    callable that accepts all three arguments.
88

89
    This middleware bridges the two interfaces, allowing ASGI2 applications
90
    to run on ASGI3 servers like uvicorn.
91
    """
92

93
    def __init__(self, app: ASGI2Application) -> None:
94
        """
95
        Initialize ASGI2 middleware.
96

97
        Args:
98
            app: ASGI2 application class (not instance)
99
                 The app should have __init__(scope) and async __call__(receive, send)
100
        """
101

102
    async def __call__(
103
        self,
104
        scope: Scope,
105
        receive: ASGIReceiveCallable,
106
        send: ASGISendCallable,
107
    ) -> None:
108
        """
109
        Run ASGI2 application as ASGI3.
110

111
        Args:
112
            scope: ASGI connection scope
113
            receive: Receive callable for incoming messages
114
            send: Send callable for outgoing messages
115

116
        This method:
117
        1. Instantiates the ASGI2 app with the scope
118
        2. Calls the instance with receive and send
119
        """
120
```
121

122
### WSGI Middleware
123

124
Adapter to run WSGI applications in ASGI.
125

126
```python { .api }
127
class WSGIMiddleware:
128
    """
129
    Adapter to run WSGI applications in ASGI.
130

131
    Bridges the synchronous WSGI interface with the asynchronous ASGI
132
    interface by running WSGI applications in a thread pool executor.
133

134
    Note: This is uvicorn's implementation. The a2wsgi package provides
135
    a more feature-complete WSGI adapter and is recommended for production use.
136
    """
137

138
    def __init__(self, app: WSGIApp, workers: int = 10) -> None:
139
        """
140
        Initialize WSGI middleware.
141

142
        Args:
143
            app: WSGI application callable
144
                 Should have signature: app(environ, start_response)
145
            workers: Number of worker threads for handling WSGI calls (default: 10)
146
                    Each WSGI request runs in a thread from this pool since
147
                    WSGI is synchronous while ASGI is asynchronous.
148
        """
149

150
    async def __call__(
151
        self,
152
        scope: Scope,
153
        receive: ASGIReceiveCallable,
154
        send: ASGISendCallable,
155
    ) -> None:
156
        """
157
        Run WSGI application in ASGI context.
158

159
        Args:
160
            scope: ASGI connection scope (only HTTP supported)
161
            receive: Receive callable for incoming messages
162
            send: Send callable for outgoing messages
163

164
        This method:
165
        1. Receives the HTTP request body
166
        2. Builds WSGI environ from ASGI scope
167
        3. Runs WSGI app in thread pool
168
        4. Sends WSGI response back through ASGI send
169
        """
170
```
171

172
### Message Logger Middleware
173

174
Middleware that logs all ASGI messages for debugging.
175

176
```python { .api }
177
class MessageLoggerMiddleware:
178
    """
179
    Middleware that logs all ASGI messages at TRACE level.
180

181
    Useful for debugging ASGI applications by showing the exact message
182
    flow between server and application. Messages with large bodies are
183
    logged with size placeholders instead of full content.
184
    """
185

186
    def __init__(self, app: ASGI3Application) -> None:
187
        """
188
        Initialize message logger middleware.
189

190
        Args:
191
            app: ASGI application to wrap
192

193
        Attributes:
194
            task_counter: Counter for assigning unique task IDs
195
            app: Wrapped ASGI application
196
            logger: Logger instance for message logging
197
        """
198

199
    async def __call__(
200
        self,
201
        scope: Scope,
202
        receive: ASGIReceiveCallable,
203
        send: ASGISendCallable,
204
    ) -> None:
205
        """
206
        Run application with message logging.
207

208
        Args:
209
            scope: ASGI connection scope
210
            receive: Receive callable for incoming messages
211
            send: Send callable for outgoing messages
212

213
        This method wraps receive and send callables to log all messages
214
        at TRACE level. Each connection is assigned a unique task ID for
215
        tracking messages across the connection lifecycle.
216
        """
217
```
218

219
### WSGI Support Types
220

221
Type definitions for WSGI applications.
222

223
```python { .api }
224
# WSGI environ dictionary
225
Environ = MutableMapping[str, Any]
226

227
# WSGI exception info tuple
228
ExcInfo = tuple[type[BaseException], BaseException, Optional[types.TracebackType]]
229

230
# WSGI start_response callable
231
StartResponse = Callable[[str, Iterable[tuple[str, str]], Optional[ExcInfo]], None]
232

233
# WSGI application callable
234
WSGIApp = Callable[[Environ, StartResponse], Union[Iterable[bytes], BaseException]]
235
```
236

237
### WSGI Helper Functions
238

239
```python { .api }
240
def build_environ(
241
    scope: HTTPScope,
242
    message: ASGIReceiveEvent,
243
    body: io.BytesIO,
244
) -> Environ:
245
    """
246
    Build WSGI environ dictionary from ASGI scope.
247

248
    Args:
249
        scope: ASGI HTTP scope
250
        message: ASGI receive event
251
        body: Request body as BytesIO
252

253
    Returns:
254
        WSGI environ dictionary with all required and optional keys
255

256
    The environ includes:
257
    - CGI variables (REQUEST_METHOD, PATH_INFO, QUERY_STRING, etc.)
258
    - Server variables (SERVER_NAME, SERVER_PORT, SERVER_PROTOCOL)
259
    - WSGI variables (wsgi.version, wsgi.url_scheme, wsgi.input, wsgi.errors)
260
    - HTTP headers (converted from ASGI format)
261
    """
262
```
263

264
## Usage Examples
265

266
### Enable Proxy Headers
267

268
```python
269
from uvicorn import run
270
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
271

272
async def app(scope, receive, send):
273
    # Your ASGI application
274
    ...
275

276
# Wrap with proxy headers middleware
277
app_with_proxy = ProxyHeadersMiddleware(
278
    app,
279
    trusted_hosts="127.0.0.1,10.0.0.0/8",
280
)
281

282
# Run with middleware
283
run(app_with_proxy, host="0.0.0.0", port=8000)
284
```
285

286
### Automatic Proxy Headers with uvicorn.run
287

288
```python
289
import uvicorn
290

291
# Proxy headers middleware is automatically applied when proxy_headers=True
292
uvicorn.run(
293
    "myapp:app",
294
    host="0.0.0.0",
295
    port=8000,
296
    proxy_headers=True,  # Enable proxy headers (default: True)
297
    forwarded_allow_ips="127.0.0.1,192.168.0.0/16",  # Trusted proxies
298
)
299
```
300

301
### Trust All Proxies (Development Only)
302

303
```python
304
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
305

306
# Trust all proxies - NOT RECOMMENDED IN PRODUCTION
307
app = ProxyHeadersMiddleware(
308
    app,
309
    trusted_hosts="*",  # Trust all sources
310
)
311
```
312

313
### Run ASGI2 Application
314

315
```python
316
from uvicorn.middleware.asgi2 import ASGI2Middleware
317

318
# ASGI2 application (class-based)
319
class MyASGI2App:
320
    def __init__(self, scope):
321
        self.scope = scope
322

323
    async def __call__(self, receive, send):
324
        assert self.scope['type'] == 'http'
325
        await send({
326
            'type': 'http.response.start',
327
            'status': 200,
328
            'headers': [[b'content-type', b'text/plain']],
329
        })
330
        await send({
331
            'type': 'http.response.body',
332
            'body': b'Hello from ASGI2!',
333
        })
334

335
# Wrap with ASGI2 middleware
336
app = ASGI2Middleware(MyASGI2App)
337

338
# Run with uvicorn
339
import uvicorn
340
uvicorn.run(app, host="127.0.0.1", port=8000)
341
```
342

343
### Automatic ASGI2 Detection
344

345
```python
346
import uvicorn
347

348
# Uvicorn automatically detects ASGI2 applications
349
# No need to manually wrap with ASGI2Middleware
350

351
class MyASGI2App:
352
    def __init__(self, scope):
353
        self.scope = scope
354

355
    async def __call__(self, receive, send):
356
        ...
357

358
uvicorn.run(
359
    MyASGI2App,
360
    host="127.0.0.1",
361
    port=8000,
362
    interface="asgi2",  # Or use "auto" for automatic detection
363
)
364
```
365

366
### Run WSGI Application
367

368
```python
369
from uvicorn.middleware.wsgi import WSGIMiddleware
370

371
# WSGI application (e.g., Flask)
372
def wsgi_app(environ, start_response):
373
    status = '200 OK'
374
    headers = [('Content-Type', 'text/plain')]
375
    start_response(status, headers)
376
    return [b'Hello from WSGI!']
377

378
# Wrap with WSGI middleware
379
app = WSGIMiddleware(wsgi_app, workers=10)
380

381
# Run with uvicorn
382
import uvicorn
383
uvicorn.run(app, host="127.0.0.1", port=8000)
384
```
385

386
### Run Flask with Uvicorn
387

388
```python
389
from flask import Flask
390
from uvicorn.middleware.wsgi import WSGIMiddleware
391
import uvicorn
392

393
# Create Flask app
394
flask_app = Flask(__name__)
395

396
@flask_app.route("/")
397
def hello():
398
    return "Hello from Flask on Uvicorn!"
399

400
# Wrap Flask with WSGI middleware
401
app = WSGIMiddleware(flask_app)
402

403
# Run with uvicorn
404
uvicorn.run(app, host="127.0.0.1", port=8000)
405
```
406

407
### Automatic WSGI Detection
408

409
```python
410
import uvicorn
411
from flask import Flask
412

413
flask_app = Flask(__name__)
414

415
@flask_app.route("/")
416
def hello():
417
    return "Hello, World!"
418

419
# Uvicorn automatically detects WSGI applications
420
uvicorn.run(
421
    flask_app,
422
    host="127.0.0.1",
423
    port=8000,
424
    interface="wsgi",  # Or use "auto" for automatic detection
425
)
426
```
427

428
### Enable Message Logging
429

430
```python
431
from uvicorn.middleware.message_logger import MessageLoggerMiddleware
432
from uvicorn.logging import TRACE_LOG_LEVEL
433
import logging
434

435
# Enable TRACE logging
436
logging.addLevelName(TRACE_LOG_LEVEL, "TRACE")
437
logging.basicConfig(level=TRACE_LOG_LEVEL)
438

439
async def app(scope, receive, send):
440
    # Your ASGI application
441
    ...
442

443
# Wrap with message logger
444
app_with_logging = MessageLoggerMiddleware(app)
445

446
# Run with uvicorn
447
import uvicorn
448
uvicorn.run(
449
    app_with_logging,
450
    host="127.0.0.1",
451
    port=8000,
452
    log_level="trace",  # Enable trace logging
453
)
454
```
455

456
### Combine Multiple Middleware
457

458
```python
459
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
460
from uvicorn.middleware.message_logger import MessageLoggerMiddleware
461

462
async def app(scope, receive, send):
463
    # Your ASGI application
464
    ...
465

466
# Apply middleware in order (innermost first)
467
app = MessageLoggerMiddleware(app)  # Applied last
468
app = ProxyHeadersMiddleware(app, trusted_hosts="127.0.0.1")  # Applied first
469

470
# Run with uvicorn
471
import uvicorn
472
uvicorn.run(app, host="0.0.0.0", port=8000)
473
```
474

475
### Custom Trusted Hosts
476

477
```python
478
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
479

480
async def app(scope, receive, send):
481
    ...
482

483
# Trust specific IP addresses
484
app = ProxyHeadersMiddleware(
485
    app,
486
    trusted_hosts=["127.0.0.1", "192.168.1.1", "192.168.1.2"],
487
)
488

489
# Trust CIDR networks
490
app = ProxyHeadersMiddleware(
491
    app,
492
    trusted_hosts=[
493
        "10.0.0.0/8",      # Private network
494
        "172.16.0.0/12",   # Private network
495
        "192.168.0.0/16",  # Private network
496
    ],
497
)
498

499
# Mix IPs and networks
500
app = ProxyHeadersMiddleware(
501
    app,
502
    trusted_hosts="127.0.0.1,10.0.0.0/8,192.168.1.100",
503
)
504
```
505

506
### WSGI with Custom Thread Pool
507

508
```python
509
from uvicorn.middleware.wsgi import WSGIMiddleware
510

511
def wsgi_app(environ, start_response):
512
    # Blocking WSGI application
513
    import time
514
    time.sleep(0.1)  # Simulate blocking I/O
515

516
    status = '200 OK'
517
    headers = [('Content-Type', 'text/plain')]
518
    start_response(status, headers)
519
    return [b'Done!']
520

521
# Increase workers for blocking applications
522
app = WSGIMiddleware(
523
    wsgi_app,
524
    workers=50,  # More threads for handling blocking calls
525
)
526

527
import uvicorn
528
uvicorn.run(app, host="127.0.0.1", port=8000)
529
```
530

531
### Inspect Proxy Headers
532

533
```python
534
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
535

536
async def app(scope, receive, send):
537
    # Access the modified scope
538
    client_host, client_port = scope['client']
539
    scheme = scope['scheme']
540

541
    body = f"Client: {client_host}:{client_port}\nScheme: {scheme}".encode()
542

543
    await send({
544
        'type': 'http.response.start',
545
        'status': 200,
546
        'headers': [[b'content-type', b'text/plain']],
547
    })
548
    await send({
549
        'type': 'http.response.body',
550
        'body': body,
551
    })
552

553
# Wrap with proxy headers
554
app = ProxyHeadersMiddleware(app, trusted_hosts="*")
555

556
# Test with curl:
557
# curl -H "X-Forwarded-For: 203.0.113.1" -H "X-Forwarded-Proto: https" http://localhost:8000
558
```
559

560
### Debug ASGI Messages
561

562
```python
563
from uvicorn.middleware.message_logger import MessageLoggerMiddleware
564
from uvicorn.logging import TRACE_LOG_LEVEL
565
import logging
566

567
# Configure trace logging
568
logging.addLevelName(TRACE_LOG_LEVEL, "TRACE")
569
logger = logging.getLogger("uvicorn")
570
logger.setLevel(TRACE_LOG_LEVEL)
571

572
handler = logging.StreamHandler()
573
handler.setLevel(TRACE_LOG_LEVEL)
574
logger.addHandler(handler)
575

576
async def app(scope, receive, send):
577
    message = await receive()  # Logged
578
    await send({  # Logged
579
        'type': 'http.response.start',
580
        'status': 200,
581
        'headers': [[b'content-type', b'text/plain']],
582
    })
583
    await send({  # Logged
584
        'type': 'http.response.body',
585
        'body': b'Hello!',
586
    })
587

588
# Wrap with message logger
589
app = MessageLoggerMiddleware(app)
590

591
import uvicorn
592
uvicorn.run(app, host="127.0.0.1", port=8000, log_level=TRACE_LOG_LEVEL)
593
```
594