ASGI stands for Asynchronous Server Gateway Interface. It is a successor to Web Server Gateway Interface (WSGI). It adds support for async handlers. Both ASGI and WSGI servers act as middleware, a piece of technology that sits in between two other technologies. In this case, it sits between a web server such as NGINX or Apache, which act as a reverse proxy, and a web framework such as Django, FastAPI, or Flask.

A reverse proxy sits in front of your backend web application and provides a whole host of security and performance features that I won't get into. ASGI and WSGI provide a common interface for web frameworks to communicate with reverse proxies. In this way, you can mix and match ASGI implementations and web frameworks. All the ASGI middleware cares about is that your web framework speaks ASGI, and all your web framework cares about is that your ASGI middleware speaks ASGI.

In this post, we implement a minimal ASGI web application and a minimal ASGI middleware. We assume familiarity with asyncio in Python, i.e., how to use async/await syntax, and some basic understanding of sockets. For more info, see the excellent https://docs.python.org/3/howto/sockets.html and https://docs.python.org/3/library/asyncio.html#module-asyncio. The Python docs are your best friend.

We build everything from scratch, without the use of any third-party packages like uvicorn.

This is more of a "learn-by-doing" post and series. I probably won't cover everything that in the ASGI spec, but I'm hoping to give you a good understanding of how things work. To learn more, visit the ASGI spec at: https://asgi.readthedocs.io/en/latest/specs/main.html.

An ASGI Application

Before we write our ASGI middleware, let's go over a simple ASGI application that uses said middleware.

# app.py
# every ASGI application has the same function signature:
async def app(scope, receive, send):
    """The top-level ASGI application.

    Args:
        scope: Information about the request.
        receive: A callable to receive events.
        send: A callable to send events.
    """
    # scope type can also be configured to something like a WebSocket,
    # so we want to make sure that it's 'http' for our purposes
    if scope['type'] != 'http':
        raise ValueError('scope type must be http')

    # the first type of event we should receive is an 'http.request'
    # let's just make sure though
    # here, we are waiting for our first event
    # we are pretty much going to ignore it though for now
    event = await receive()
    if event['type'] != 'http.request':
        raise ValueError('event type must be http.request')

    # let's setup our response body and headers
    # in this case, it's a really simple response, but
    # pretend that Django/Flask/FastAPI is doing something
    # important here
    response = b'Hello, world!'
    # we need to set the headers too so that the client
    # knows how to interpret the response
    headers = [(b'content-type', b'tex
t/plain')]

    # now, we can finally start sending a response
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': headers
    })

    await send({
        'type': 'http.response.body',
        'body': response
    })

Notice that at the very end of this snippet, we send out the status and headers as a different send event than the body itself. This is similar to how HTTP itself works, with the status-code and headers coming first in a response.

ASGI is meant to not block the main thread. Thus, one can send multiple http.response.body events before finally completing the response. One can do this by using the more_body key in the dictionary like so:

await send({
    'type': 'http.response.body',
    'body': b'Hello, ',
    'more_body': True  # new
})

await send({
    'type': 'http.response.body',
    'body': b'world!',
    'more_body': False  # new

This allows one to chunk up responses and stream them out rather than sending one long, blocking response.

We can now run this with something like Daphne or Uvicorn. I'd hesitate to even call this a "web framework"! It is still missing a ton of features like request routing, etc., but we'll get to that later.

Let's try running it with uvicorn:

$ uvicorn app:app
INFO:     Started server process [88244]
INFO:     Waiting for application startup.
INFO:     ASGI 'lifespan' protocol appears unsupported.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     127.0.0.1:55692 - "GET / HTTP/1.1" 200 OK
INFO:     127.0.0.1:55692 - "GET /favicon.ico HTTP/1.1" 200 OK

Open up localhost:8000:

Success! Now onto the server itself.

An ASGI server

This gets a little trickier now. We need to do things like open up a socket and handle HTTP. I will assume some familiarity with those topics.

# server.py
import asyncio
import socket

# TODO in a real ASGI server we wouldn't hardcode this
from app import app


async def handle_client(client_socket, app):
    # create a stream reader and writer for the client
    # these are abstractions over the socket connection that provide
    # buffering and better compatibility with the asyncio ecosystem
    reader, writer = await asyncio.open_connection(sock=client_socket)

    # ASGI scope for the HTTP connection
    # used by the ASGI app
    scope = {
        'type': 'http',
        'http_version': '1.1',
        'method': 'GET',
        'path': '/',
        'headers': [],
    }

    async def receive():
        """Used by the ASGI app."""
        # TODO In a real server, you'd parse the HTTP request here
        request = await reader.read(1024)
        return {
            'type': 'http.request',
            'body': b'',
        }

    async def send(message):
        """Used by the ASGI app."""
        if message['type'] == 'http.response.start':
            status_line = f"HTTP/1.1 {message['status']} OK\r\n"
            headers = "".join(
                f"{name.decode()}: {value.decode()}\r\n" for name, value in message["headers"]
            )
            writer.write(status_line.encode() + headers.encode() + b"\r\n")
            # ensure that the data is sent, wait until the buffer is
            # sufficiently emptied, handle backpressure
            await writer.drain()
        elif message['type'] == 'http.response.body':
            writer.write(message['body'])

            if not message.get('more_body', False):
                # indicate the end of the response
                writer.write_eof()

            try:
                await writer.drain()
            except ConnectionError:
                # this occurs in browsers for some reason
                # but not curl
                print('connection was reset by the client.')
                return

            # close the stream and the underlying socket
            writer.close()
            # wait until the stream is closed to ensure that
            # all data has been flushed before exiting
            await writer.wait_closed()

    # call the ASGI application
    await app(scope, receive, send)

async def main(host='127.0.0.1', port=8000):
    # create a listening socket
    # IPv4, TCP socket
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # allow reuse of a port, which is useful in development
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind((host, port))
    server_socket.listen()
    # see more at https://docs.python.org/3/howto/sockets.html

    print(f"Serving on {host}:{port}")

    # we need a reference to the loop later to asynchronously
    # accept connections
    loop = asyncio.get_event_loop()
    # don't block on socket reads or writes
    # this is importang for asynchronous communication
    server_socket.setblocking(False)
    while True:
        # asynchronously accept a connection
        client_socket, _ = await loop.sock_accept(server_socket)
        # call the ASGI application app in a coroutine
        # TODO in something like uvicorn the app would be dynamically
        # configured via the cli, but here we hardcode the import and
        # function call for convenience
        loop.create_task(handle_client(client_socket, app))

# Run the server
asyncio.run(main())

We can finally run the server:

$ python3 server.py 
Serving on 127.0.0.1:8000

Here is another way to read from the server:

$ curl localhost:8000
Hello, world!

Conclusion

If this all seems complicated, that's because it is. This is what web frameworks and middleware is good for. It does the work for you. Still, it's good to know how things work to appreciate what you don't have to do every time you want to write a web application. This is a really meaty subject, so please ask questions. I am learning all of this myself as well.