Making a Simple HTTP Server with Asyncio Protocols

Posted on July 1, 2025

Asyncio has two main abstractions for handling network connections – Streams and Protocols. Asyncio Streams are a nice high-level primitive for handling TCP connections with some very convenient helper methods such as readline, readuntil, and readexactly. Protocols, on the other hand, are a lower-level callback-based approach for handling socket connections. They are more complex than Streams, but are faster, since they avoid the overhead of coroutines and can react to events more immediately via callbacks.

In this article, I’m going to talk about how you can use these Protocols to make a simple HTTP server that can accept GET requests and return data for specific routes defined from functions via a route decorator.

Connection Object

To start making our HTTP server, the first thing to do is create a class that inherits from Asyncio.Protocol. This ConnectionHandler object will be used to handle all of the client connections – one instance per connection. This connection handler object will have a few standardized methods that the Async Protocol will call in response to certain events such as when a connection is made, closed, or data is received from the client.

Each instance of this connection handler will need three instance variables to keep track of each client connection state:

routes: This variable is going to store the mappings between the URL endpoints and the functions that will eventually create the output for each route.
buffer: This bytearray is going to be used to store the data received from the data_received callback method. It’s important to have some sort of buffer because there is no guarantee of how much data data_received will give you on each callback, so we need to keep track of the data received from the client in case the HTTP request arrives in multiple chunks.
transport: This is just a placeholder for now, but when the connection is made, the Async Protocol will give us a transport object that will let us talk to the client.

import asyncio


class ConnectionHandler(asyncio.Protocol):
   def __init__(self, routes):
       self.routes = routes
       self.buffer = bytearray()
       self.transport = None

Handling Connections

The next method in the connection handler class to make is connection_made, which is automatically called on a new connection. All we need this method to do for our simple server is to save the transport object to an instance variable so that we can send data back to the client later. In a more complex application, there may be more logging here – for example, you could use the transport method get_extra_info to get and track the IP address of clients who connect.

def connection_made(self, transport):
   self.transport = transport

HTTP Parsing

The next, and most complicated method to tackle is data_received, which is the callback that’s triggered every time a new chunk of data is received from the client.

This will be the method that handles most of the core logic, including parsing the HTTP requests. For parsing client requests, it’s important to understand the HTTP spec and how each request is formatted. The first line of an HTTP request tells you what the method is (GET, PUT, POST, etc), the endpoint, and the HTTP version. Then, each subsequent line is an HTTP header to give more metadata to the server. Each line is separated by a \r\n and the last header line ends with \r\n\r\n to split the headers from the body of the request. This is useful to know, as we can try to find \r\n\r\n in our received data (self.buffer) to determine when we’ve received all of the HTTP headers.

Before parsing anything, the first thing this method needs to do is add the new bytes to our bytearray object. Then, we need to see if all of the header data has been received by trying to find \r\n\r\n:

def data_received(self, data):
   self.buffer.extend(data)

   request_headers_index = self.buffer.find(b'\r\n\r\n')
   if request_headers_index == -1:
       return
   …

For this server, since I'm only going to write it to handle GET requests, we don’t care about the body of a request, only the HTTP headers, so the next line of this method cuts the string off where the headers end (just in case a request comes in with a body). After making sure that we just have the headers of a request, we need to format them into a more workable format:

def data_received(self, data):
   …
   request_headers = self.buffer[:request_headers_index].decode(errors='ignore')
   request = ConnectionHandler.parse_request(request_headers)
   …

@staticmethod
def parse_request(header_data):
   request = {}

   lines = header_data.split('\r\n')

   request_line = lines[0]
   method, path, _ = request_line.split()
   request['method'] = method
   request['path'] = path

   for line in lines[1:]:
       if ':' in line:
           key, value = line.split(':', maxsplit=1)
           request[key.strip()] = value.strip()

   return request

With all of that code, if the client sends a request like this:

GET /home HTTP/1.1\r\n
Host: localhost:9999\r\n
User-Agent: curl/8.0.1\r\n
Accept: */*\r\n
\r\n

We’ll get a nicely formatted dictionary with all of the relevant info from parse_request like such:

{
    'method': 'GET',
    'path': '/home',
    'Host': 'localhost:9999',
    'User-Agent': 'curl/8.0.1',
    'Accept': '*/*'
}

Sending an HTTP Response

Once we’ve parsed the HTTP request, we can finally send a response back to the client!

If the request isn’t a GET request or the path doesn’t have a corresponding handler function in self.routes, we’ll just return an error code. Otherwise, we can grab the handler function from self.routes and pass it to send_response, which will await the handler function to get the response body.

Since data_received is just a regular callback method, we can’t await the handler coroutine inside of it. That’s why I’m using the send_response coroutine, which will then await the handler coroutine. To get send_response to run asynchronously, we need to add it to the event loop via the create_task function.

This does introduce a small delay because the response doesn’t get sent until the event loop gets to the send_response coroutine. Also, in this simple example, our handlers don’t actually need to be asynchronous — they're just going to be returning static HTML — but in a real server where these handlers might hit a database or make API calls, you'd definitely want that async support. So, even though this pattern is a bit slower, it gives us flexibility to have async handler functions, which is pretty important.

def data_received(self, data):
  
   # Data parsing code ...
  
   if request['method'] != 'GET':
       response_coro = self.send_response(405, "<h1>405 Method Not Allowed</h1>")
   elif request['path'] not in self.routes:
       response_coro = self.send_response(404, "<h1>404 Not Found</h1>")
   else:
       handler = self.routes[request['path']]
       response_coro = self.send_response(200, handler)

   asyncio.create_task(response_coro)

For the send_response method, I wrote it to accept either a coroutine function or plain strings, since the error responses are just standard static strings. Once we get the response body, the method builds a super simple HTTP response with just the essentials: Content-Type, Content-Length, and Connection. After making the response, we can send it to the client via the transport’s write method and then close the connection with the client with close.

async def send_response(self, status_code, content=None):
   if inspect.iscoroutinefunction(content):
       body = await content()
   else:
       body = content

   response = (
       f"HTTP/1.1 {status_code} {HTTPStatus(status_code).phrase}\r\n"
       "Content-Type: text/html\r\n"
       f"Content-Length: {len(body.encode())}\r\n"
       f"Connection: close\r\n"
       f"\r\n"
       f"{body}"
   )

   self.transport.write(response.encode())
   self.transport.close()

A Route Decorator

Even though this isn’t really necessary for the point of this article (Asyncio Protocols), I think it’s pretty interesting to see how packages like FastAPI use the Python decorator syntax to make a decorator that just logs functions to handle specific routes.

Normally, with a decorator, you want to wrap the original function to achieve some sort of extra functionality, but since we just want to log specific functions, we can just return the exact same function and store the first-class version of the functions in a route dictionary (self.routes).

It’s actually that simple! Below is the route decorator with self.routes to store all of the handler functions:

class HTTPServer:
   def __init__(self):
       self.routes = {}

   def route(self, endpoint: str):
       def decorator(func):
           self.routes[endpoint] = func
           return func
       return decorator

With this decorator, we can then create functions like the ones below to handle specific routes. I know, I know, technically these could be regular functions, but I wanted to make this article pretty general, and if you’re using an asynchronous HTTP server, there’s a good chance that you will also want your route handlers to be async.

app = HTTPServer()

@app.route('/home')
async def home():
   return (
       '<h1>Welcome to my Home Page</h1>'
       '<a href="/contact">Contact us</a>'
   )

@app.route('/contact')
async def contact():
   return (
       '<h1>Contact Page</h1>'
       '<p>Email: articles@jacobpadilla.com</p>'
       '<a href="/home">Back to Home</a>'
   )

Putting It All Together

Now to put all of the pieces together! To actually initialize the server, we need to use the Asyncio create_server method, which will bind the server to a port and create a new instance of, in this case, ConnectionHandler when a new client tries to connect to the server. Lastly, I created a run method which can be called synchronously to kick off the async portion of the code.

class HTTPServer:
    ...

    def run(self):
        asyncio.run(self.serve())

    async def serve(self, port: int = 9999):
        loop = asyncio.get_running_loop()

        server = await loop.create_server(
            lambda: ConnectionHandler(self.routes),
            '127.0.0.1', port
        )

        async with server:
            await server.serve_forever()

And all put together, we get this! A simple HTTP server that can accept GET requests and return data for specific routes.

import asyncio
import inspect
import logging
from http import HTTPStatus


logging.basicConfig(
   level=logging.INFO,
   format='[%(asctime)s] [%(levelname)s] %(message)s',
   datefmt='%H:%M:%S'
)
logger = logging.getLogger(__name__)


class ConnectionHandler(asyncio.Protocol):
   def __init__(self, routes):
       self.routes = routes
       self.buffer = bytearray()
       self.transport = None

   def connection_made(self, transport):
       self.transport = transport

   def data_received(self, data):
       self.buffer.extend(data)

       request_headers_index = self.buffer.find(b'\r\n\r\n')
       if request_headers_index == -1:
           return

       request_headers = self.buffer[:request_headers_index].decode(errors='ignore')
       request = ConnectionHandler.parse_request(request_headers)
       logger.info(f"{request.get('method')} {request.get('path')} - {request.get('User-Agent')}")

       if request['method'] != 'GET':
           response_coro = self.send_response(405, "<h1>405 Method Not Allowed</h1>")
       elif request['path'] not in self.routes:
           response_coro = self.send_response(404, "<h1>404 Not Found</h1>")
       else:
           handler = self.routes[request['path']]
           response_coro = self.send_response(200, handler)

       asyncio.create_task(response_coro)

   @staticmethod
   def parse_request(header_data):
       request = {}

       lines = header_data.split('\r\n')

       request_line = lines[0]
       method, path, _ = request_line.split()
       request['method'] = method
       request['path'] = path

       for line in lines[1:]:
           if ':' in line:
               key, value = line.split(':', maxsplit=1)
               request[key.strip()] = value.strip()

       return request

   async def send_response(self, status_code, content=None):
       if inspect.iscoroutinefunction(content):
           body = await content()
       else:
           body = content

       response = (
           f"HTTP/1.1 {status_code} {HTTPStatus(status_code).phrase}\r\n"
           "Content-Type: text/html\r\n"
           f"Content-Length: {len(body.encode())}\r\n"
           f"Connection: close\r\n"
           f"\r\n"
           f"{body}"
       )

       self.transport.write(response.encode())
       self.transport.close()


class HTTPServer:
   def __init__(self):
       self.routes = {}

   def route(self, endpoint: str):
       def decorator(func):
           self.routes[endpoint] = func
           return func
       return decorator

   def run(self):
       asyncio.run(self.serve())

   async def serve(self, port: int = 9999):
       loop = asyncio.get_running_loop()

       server = await loop.create_server(
           lambda: ConnectionHandler(self.routes),
           '127.0.0.1', port
       )

       async with server:
           await server.serve_forever()


app = HTTPServer()

@app.route('/home')
async def home():
   return (
       '<h1>Welcome to my Home Page</h1>'
       '<a href="/contact">Contact us</a>'
   )

@app.route('/contact')
async def contact():
   return (
       '<h1>Contact Page</h1>'
       '<p>Email: articles@jacobpadilla.com</p>'
       '<a href="/home">Back to Home</a>'
   )

app.run()

Performance

Asyncio Protocols are low-level for a reason: they’re super fast and also give you extremely fine-grained control of TCP connections. This simple HTTP server can work through around 100,000 requests in just around 4.2 seconds according to Apache Benchmark:

$ ab -n 100000 -c 100 http://127.0.0.1:9999/home

FastAPI, on the other hand, takes roughly 32 seconds to work through the 100k requests! Now this obviously isn’t a very fair test since FastAPI does SO MUCH MORE like Pydantic validation, error handling, and many more things. That being said, if you just need a dead-simple HTTP server, it’s kind of interesting to see how much faster you can make things by building your own HTTP server.

Hope this article helped you learn more about Asyncio Protocols! Thanks for reading.