aiohttp Async Web Server and Proxy

Problem Statement

Modern microservice architectures and edge API gateways require lightweight, low-latency HTTP servers that can route traffic, terminate connections, and act as dynamic reverse proxies. Standard synchronous WSGI servers struggle under high concurrency because each inbound or outbound connection consumes a dedicated worker thread or process.

This script solves the problem of high-concurrency request routing and reverse proxying by building an asynchronous HTTP server using Python's aiohttp framework. It handles non-blocking request forwarding, connection pooling, hop-by-hop header filtering, response streaming, and real-time operational health reporting without blocking the asynchronous event loop.

How It Works

The server relies on aiohttp.web to handle inbound client connections asynchronously and aiohttp.ClientSession to manage pooled outbound connections to upstream targets. The data flow follows a non-blocking streaming architecture:

1. Lifecycle Management: Upon startup, an persistent aiohttp.ClientSession with a TCPConnector is initialized and attached to the application context. This enables HTTP connection reuse across proxied requests.
2. Hop-by-Hop Header Sanitization: Before forwarding requests or returning responses, the proxy strips standard HTTP hop-by-hop headers (such as Connection, Transfer-Encoding, and Keep-Alive) to avoid breaking HTTP/1.1 protocol rules between separate connection segments.
3. Response Streaming: Proxied payloads are piped asynchronously from upstream responses down to the client via aiohttp.web.StreamResponse in chunks (iter_any()), preventing high memory usage even when handling multi-gigabyte transfers.
4. Endpoint Routing: Local routes like /health and /metrics return operational telemetry directly, while catch-all /proxy/{tail:.*} routes map incoming requests dynamically to upstream destinations.

Full Script

#!/usr/bin/env python3
"""
aiohttp Async Web Server and Proxy

A production-ready asynchronous HTTP web server and reverse proxy built with aiohttp.
Supports streaming proxies, connection pooling, lifecycle management, and health metrics.
"""

import argparse
import asyncio
import logging
import os
import sys
import time
from typing import AsyncGenerator
from aiohttp import ClientError, ClientSession, TCPConnector, web

# ------------------------------------------------------------------------------
# Configuration & Constants
# ------------------------------------------------------------------------------
DEFAULT_HOST = os.environ.get("SERVER_HOST", "0.0.0.0")
DEFAULT_PORT = int(os.environ.get("SERVER_PORT", "8080"))
DEFAULT_UPSTREAM = os.environ.get("UPSTREAM_URL", "https://httpbin.org")
DEFAULT_TIMEOUT = float(os.environ.get("REQUEST_TIMEOUT", "30.0"))
DEFAULT_MAX_CONN = int(os.environ.get("MAX_CONNECTIONS", "100"))

# Standard HTTP hop-by-hop headers that must not be forwarded by a proxy
HOP_BY_HOP_HEADERS = {
    "connection",
    "keep-alive",
    "proxy-authenticate",
    "proxy-authorization",
    "te",
    "trailers",
    "transfer-encoding",
    "upgrade",
}

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("async_proxy")


# ------------------------------------------------------------------------------
# Helper Functions
# ------------------------------------------------------------------------------
def filter_headers(headers: web.HTTPHeaders | dict[str, str]) -> dict[str, str]:
    """Remove hop-by-hop headers from request/response header maps."""
    return {
        key: value
        for key, value in headers.items()
        if key.lower() not in HOP_BY_HOP_HEADERS and key.lower() != "host"
    }


# ------------------------------------------------------------------------------
# Application Lifecycle Handlers
# ------------------------------------------------------------------------------
async def on_startup(app: web.Application) -> None:
    """Initialize persistent client session and metrics tracking on startup."""
    connector = TCPConnector(
        limit=app["config"]["max_connections"],
        ttl_dns_cache=300,
        enable_cleanup_closed=True,
    )
    timeout = ClientSession_timeout = aiohttp.ClientTimeout(
        total=app["config"]["timeout"]
    )
    
    # Store reusable client session in application context
    app["client_session"] = ClientSession(connector=connector, timeout=timeout)
    app["start_time"] = time.time()
    app["metrics"] = {
        "requests_total": 0,
        "requests_proxied": 0,
        "errors_total": 0,
    }
    logger.info(
        "Server initialized. Upstream target: %s", app["config"]["upstream"]
    )


async def on_cleanup(app: web.Application) -> None:
    """Gracefully close outbound HTTP client sessions on shutdown."""
    logger.info("Closing upstream client session...")
    session: ClientSession | None = app.get("client_session")
    if session and not session.closed:
        await session.close()
    logger.info("Cleanup complete.")


# ------------------------------------------------------------------------------
# Route Handlers
# ------------------------------------------------------------------------------
async def handle_health(request: web.Request) -> web.Response:
    """Return JSON system status and operational metrics."""
    app = request.app
    app["metrics"]["requests_total"] += 1
    
    uptime = time.time() - app["start_time"]
    payload = {
        "status": "healthy",
        "uptime_seconds": round(uptime, 2),
        "upstream": app["config"]["upstream"],
        "metrics": app["metrics"],
    }
    return web.json_response(payload)


async def handle_proxy(request: web.Request) -> web.StreamResponse:
    """
    Reverse proxy handler that forwards incoming client requests
    to the upstream target and streams back the response body.
    """
    app = request.app
    app["metrics"]["requests_total"] += 1
    app["metrics"]["requests_proxied"] += 1

    session: ClientSession = app["client_session"]
    upstream_base = app["config"]["upstream"].rstrip("/")
    
    # Construct destination URL preserving original path and query string
    tail = request.match_info.get("tail", "")
    target_url = f"{upstream_base}/{tail}"
    if request.query_string:
        target_url = f"{target_url}?{request.query_string}"

    # Prepare forwarded headers
    forwarded_headers = filter_headers(request.headers)
    forwarded_headers["X-Forwarded-For"] = request.remote or "unknown"
    forwarded_headers["X-Forwarded-Proto"] = request.scheme

    # Read client request body stream
    body_data = await request.read()

    try:
        async with session.request(
            method=request.method,
            url=target_url,
            headers=forwarded_headers,
            data=body_data,
            allow_redirects=False,
        ) as upstream_response:

            # Create downstream response matching upstream status and filtered headers
            out_headers = filter_headers(upstream_response.headers)
            response = web.StreamResponse(
                status=upstream_response.status,
                headers=out_headers,
            )

            # Prepare client response stream
            await response.prepare(request)

            # Stream payload chunks from upstream to client without loading full body in memory
            async for chunk in upstream_response.content.iter_any():
                await response.write(chunk)

            await response.write_eof()
            return response

    except ClientError as exc:
        app["metrics"]["errors_total"] += 1
        logger.error("Upstream communication failure: %s", exc)
        return web.json_response(
            {"error": "Bad Gateway", "details": str(exc)}, status=502
        )
    except asyncio.TimeoutError:
        app["metrics"]["errors_total"] += 1
        logger.error("Upstream request timed out: %s", target_url)
        return web.json_response(
            {"error": "Gateway Timeout", "details": "Upstream service timed out"},
            status=504,
        )
    except Exception as exc:
        app["metrics"]["errors_total"] += 1
        logger.exception("Unexpected proxy exception occurred")
        return web.json_response(
            {"error": "Internal Server Error", "details": str(exc)}, status=500
        )


# ------------------------------------------------------------------------------
# Application Factory & CLI Entry Point
# ------------------------------------------------------------------------------
def build_app(config: dict) -> web.Application:
    """Construct and configure the aiohttp Application instance."""
    app = web.Application()
    app["config"] = config

    # Register lifecycle hooks
    app.on_startup.append(on_startup)
    app.on_cleanup.append(on_cleanup)

    # Register routes
    app.router.add_get("/health", handle_health)
    app.router.add_route("*", "/proxy/{tail:.*}", handle_proxy)

    return app


def parse_args() -> argparse.Namespace:
    """Parse command line arguments with environment variable fallbacks."""
    parser = argparse.ArgumentParser(
        description="aiohttp Async Web Server and Reverse Proxy"
    )
    parser.add_argument(
        "--host",
        default=DEFAULT_HOST,
        help=f"Host address to bind to (default: {DEFAULT_HOST})",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=DEFAULT_PORT,
        help=f"Port to listen on (default: {DEFAULT_PORT})",
    )
    parser.add_argument(
        "--upstream",
        default=DEFAULT_UPSTREAM,
        help=f"Base upstream URL (default: {DEFAULT_UPSTREAM})",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=DEFAULT_TIMEOUT,
        help=f"Upstream request timeout in seconds (default: {DEFAULT_TIMEOUT})",
    )
    parser.add_argument(
        "--max-connections",
        type=int,
        default=DEFAULT_MAX_CONN,
        help=f"Connection pool limit (default: {DEFAULT_MAX_CONN})",
    )
    return parser.parse_args()


if __name__ == "__main__":
    import aiohttp

    args = parse_args()
    app_config = {
        "upstream": args.upstream,
        "timeout": args.timeout,
        "max_connections": args.max_connections,
    }

    application = build_app(app_config)

    logger.info("Starting web server on %s:%d", args.host, args.port)
    web.run_app(
        application,
        host=args.host,
        port=args.port,
        access_log_format='%a "%r" %s %b "%{User-Agent}i"',
    )

Configuration

The script accepts configuration via environment variables or command-line flags. Command-line flags take precedence over environment variables.

Usage Examples

1. Basic Local Server Execution

Run the server using default arguments (proxies requests to https://httpbin.org):

python3 async_server.py --host 127.0.0.1 --port 8080

Output:

2026-03-31 10:00:00,000 [INFO] async_proxy: Starting web server on 127.0.0.1:8080
2026-03-31 10:00:00,005 [INFO] async_proxy: Server initialized. Upstream target: https://httpbin.org
======== Running on http://127.0.0.1:8080 ========
(Press CTRL+C to quit)

2. Proxying Requests to Upstream

Execute requests via curl through the /proxy/ endpoint:

# Query the proxied upstream endpoint
curl -i http://127.0.0.1:8080/proxy/get?category=web

Output:

HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 312
Date: Tue, 31 Mar 2026 10:01:00 GMT
Server: gunicorn/19.9.0

{
  "args": {
    "category": "web"
  }, 
  "headers": {
    "Accept": "*/*", 
    "X-Forwarded-For": "127.0.0.1", 
    "X-Forwarded-Proto": "http"
  }, 
  "url": "https://httpbin.org/get?category=web"
}

3. Monitoring Health Telemetry

Inspect live server metrics and status:

curl -s http://127.0.0.1:8080/health | jq .

Output:

{
  "status": "healthy",
  "uptime_seconds": 42.15,
  "upstream": "https://httpbin.org",
  "metrics": {
    "requests_total": 2,
    "requests_proxied": 1,
    "errors_total": 0
  }
}

Error Handling

This script handles server failures gracefully to ensure stability under load:

• Hop-by-Hop Header Stripping: Prevents protocol corruption by filtering out low-level transfer headers (Connection, Transfer-Encoding, etc.) before forwarding requests downstream or upstream.
• Bad Gateway (502 Bad Gateway): Catches aiohttp.ClientError when upstream hosts are unreachable, DNS resolution fails, or SSL handshake fails.
• Gateway Timeout (504 Gateway Timeout): Intercepts asyncio.TimeoutError when the upstream service fails to send data within the configured timeout window.
• Unhandled Exceptions (500 Internal Server Error): Traps unexpected top-level server exceptions and logs full traceback details without crashing the primary event loop.
• Graceful Lifecycle Shutdown: Listens for termination signals (SIGINT, SIGTERM) and triggers the on_cleanup signal to close underlying socket connections via ClientSession.close().

Extending Ideas

1. Authentication Middleware: Add JWT validation or API key verification in an @web.middleware wrapper prior to reaching the route handlers.
2. Circuit Breaker Integration: Implement threshold-based circuit breaking (e.g., using pybreaker or custom logic) to immediately fail fast if the upstream service returns continuous 5xx errors.
3. Dynamic Multi-Upstream Routing: Extend the route handler to map request headers (e.g., X-Target-Backend) or URL prefix segments to multiple upstream destination pools.
4. Prometheus Telemetry Export: Replace the simple in-memory /health JSON structure with standard prometheus_client metrics export for integration with Grafana.