Build a Mini-FastAPI from Scratch: Learn ASGI & Routing Internals

Aug 06, 2025

Leapcell: The Best of Serverless Web Hosting

Building a Simplified FastAPI from Scratch: Understanding ASGI and Core Routing

Introduction: Why Reinvent This Wheel?

When we talk about Python asynchronous web frameworks, FastAPI is undoubtedly the brightest star in recent years. It has gained widespread acclaim for its impressive performance, automatic API documentation generation, and type hint support. But have you ever wondered: what magic lies behind this powerful framework?

Today, we'll build a simplified version of FastAPI from scratch, focusing on understanding two core concepts: the ASGI protocol and the routing system. By constructing it with our own hands, you'll grasp the working principles of modern asynchronous web frameworks. This won't just help you use FastAPI better—it'll enable you to quickly identify the root cause when problems arise.

What is ASGI? Why is it More Advanced than WSGI?

Before we start coding, we need to understand ASGI (Asynchronous Server Gateway Interface)—the foundation that allows FastAPI to achieve high-performance asynchronous processing.

Limitations of WSGI

If you've used Django or Flask, you've probably heard of WSGI (Web Server Gateway Interface). WSGI is a synchronous interface specification between Python web applications and servers, but it has obvious flaws:

Can only handle one request at a time, no concurrency
Doesn't support long-lived connections (like WebSocket)
Can't fully leverage the advantages of asynchronous I/O

Advantages of ASGI

ASGI was created to solve these problems:

Fully asynchronous, supporting concurrent processing of multiple requests
Compatible with WebSocket and HTTP/2
Allows middleware to work in asynchronous environments
Supports asynchronous events throughout the request lifecycle

Simply put, ASGI defines a standard interface that allows asynchronous web applications to communicate with servers (like Uvicorn). Next, we'll implement a minimalist ASGI server.

Step 1: Implement a Basic ASGI Server

An ASGI application is essentially a callable object (function or class) that receives three parameters: scope, receive, and send.

# asgi_server.py
import socket
import asyncio
import json
from typing import Callable, Awaitable, Dict, Any

# ASGI application type definition
ASGIApp = Callable[[Dict[str, Any], Callable[[], Awaitable[Dict]]], Awaitable[None]]

class ASGIServer:
    def __init__(self, host: str = "127.0.0.1", port: int = 8000):
        self.host = host
        self.port = port
        self.app: ASGIApp = self.default_app  # Default application

    async def default_app(self, scope: Dict[str, Any], receive: Callable, send: Callable):
        """Default application: returns 404 response"""
        if scope["type"] == "http":
            await send({
                "type": "http.response.start",
                "status": 404,
                "headers": [(b"content-type", b"text/plain")]
            })
            await send({
                "type": "http.response.body",
                "body": b"Not Found"
            })

    async def handle_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        """Handles new connections, parses HTTP requests and passes to ASGI application"""
        data = await reader.read(1024)
        request = data.decode().split("\r\n")
        method, path, _ = request[0].split()

        # Build ASGI scope
        scope = {
            "type": "http",
            "method": method,
            "path": path,
            "headers": []
        }

        # Parse request headers
        for line in request[1:]:
            if line == "":
                break
            key, value = line.split(":", 1)
            scope["headers"].append((key.strip().lower().encode(), value.strip().encode()))

        # Define receive and send methods
        async def receive() -> Dict:
            """Simulates receiving messages (simplified version)"""
            return {"type": "http.request", "body": b""}

        async def send(message: Dict):
            """Sends response to client"""
            if message["type"] == "http.response.start":
                status = message["status"]
                status_line = f"HTTP/1.1 {status} OK\r\n"
                headers = "".join([f"{k.decode()}: {v.decode()}\r\n" for k, v in message["headers"]])
                writer.write(f"{status_line}{headers}\r\n".encode())

            if message["type"] == "http.response.body":
                writer.write(message["body"])
                await writer.drain()
                writer.close()

        # Call ASGI application
        await self.app(scope, receive, send)

    async def run(self):
        """Starts the server"""
        server = await asyncio.start_server(
            self.handle_connection, self.host, self.port
        )
        print(f"Server running on http://{self.host}:{self.port}")
        async with server:
            await server.serve_forever()

# Run the server
if __name__ == "__main__":
    server = ASGIServer()
    asyncio.run(server.run())

This simplified ASGI server can handle basic HTTP requests and return responses. Test it out: after running the script, visit

http://127.0.0.1:8000

and you'll see "Not Found" because we haven't defined any routes yet.

Step 2: Implement the Routing System

One of FastAPI's most intuitive features is its elegant route definition, like:

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

Let's implement similar routing functionality.

Routing Core Component Design

We need three core components:

Router: Manages all routing rules
Decorators: @get, @post, etc., for registering routes
Path matching: Handles dynamic path parameters (like /items/{item_id})

# router.py
from typing import Callable, Awaitable, Dict, Any, List, Tuple, Pattern
import re
from functools import wraps

# Route type definition
RouteHandler = Callable[[Dict[str, Any]], Awaitable[Dict[str, Any]]]

class Route:
    def __init__(self, path: str, methods: List[str], handler: RouteHandler):
        self.path = path
        self.methods = [m.upper() for m in methods]
        self.handler = handler
        self.path_pattern, self.param_names = self.compile_path(path)

    def compile_path(self, path: str) -> Tuple[Pattern, List[str]]:
        """Converts path to regular expression and extracts parameter names"""
        param_names = []
        pattern = re.sub(r"{(\w+)}", lambda m: (param_names.append(m.group(1)), r"(\w+)")[1], path)
        return re.compile(f"^{pattern}$"), param_names

    def match(self, path: str, method: str) -> Tuple[bool, Dict[str, Any]]:
        """Matches path and method, returns parameters"""
        if method not in self.methods:
            return False, {}

        match = self.path_pattern.match(path)
        if not match:
            return False, {}

        params = dict(zip(self.param_names, match.groups()))
        return True, params

class Router:
    def __init__(self):
        self.routes: List[Route] = []

    def add_route(self, path: str, methods: List[str], handler: RouteHandler):
        """Adds a route"""
        self.routes.append(Route(path, methods, handler))

    def route(self, path: str, methods: List[str]):
        """Route decorator"""
        def decorator(handler: RouteHandler):
            self.add_route(path, methods, handler)
            @wraps(handler)
            async def wrapper(*args, **kwargs):
                return await handler(*args, **kwargs)
            return wrapper
        return decorator

    # Shortcut methods
    def get(self, path: str):
        return self.route(path, ["GET"])

    def post(self, path: str):
        return self.route(path, ["POST"])

    async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
        """Handles requests, finds matching route and executes it"""
        path = scope["path"]
        method = scope["method"]

        for route in self.routes:
            matched, params = route.match(path, method)
            if matched:
                # Parse query parameters
                query_params = self.parse_query_params(scope)
                # Merge path parameters and query parameters
                request_data = {** params, **query_params}
                # Call handler function
                return await route.handler(request_data)

        # No route found
        return {"status": 404, "body": {"detail": "Not Found"}}

    def parse_query_params(self, scope: Dict[str, Any]) -> Dict[str, Any]:
        """Parses query parameters (simplified version)"""
        # In actual ASGI, query parameters are in scope["query_string"]
        query_string = scope.get("query_string", b"").decode()
        params = {}
        if query_string:
            for pair in query_string.split("&"):
                if "=" in pair:
                    key, value = pair.split("=", 1)
                    params[key] = value
        return params

Integrating Routing with the ASGI Server

Now we need to modify our ASGI server to use our routing system:

# Add routing support to ASGIServer class
class ASGIServer:
    def __init__(self, host: str = "127.0.0.1", port: int = 8000):
        self.host = host
        self.port = port
        self.router = Router()  # Instantiate router
        self.app = self.asgi_app  # Use routing-enabled ASGI application

    async def asgi_app(self, scope: Dict[str, Any], receive: Callable, send: Callable):
        """ASGI application with routing functionality"""
        if scope["type"] == "http":
            # Handle request
            response = await self.router.handle(scope, receive)
            status = response.get("status", 200)
            body = json.dumps(response.get("body", {})).encode()

            # Send response
            await send({
                "type": "http.response.start",
                "status": status,
                "headers": [(b"content-type", b"application/json")]
            })
            await send({
                "type": "http.response.body",
                "body": body
            })

Step 3: Implement Parameter Parsing and Type Conversion

One of FastAPI's highlights is its automatic parameter parsing and type conversion. Let's implement this feature:

# Add type conversion to Router's handle method
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
    # ... previous code ...

    if matched:
        # Parse query parameters
        query_params = self.parse_query_params(scope)
        # Merge path parameters and query parameters
        raw_data = {** params, **query_params}

        # Get parameter type annotations from handler function
        handler_params = route.handler.__annotations__

        # Type conversion
        request_data = {}
        for key, value in raw_data.items():
            if key in handler_params:
                target_type = handler_params[key]
                try:
                    # Attempt type conversion
                    request_data[key] = target_type(value)
                except (ValueError, TypeError):
                    return {
                        "status": 400,
                        "body": {"detail": f"Invalid type for {key}, expected {target_type}"}
                    }
            else:
                request_data[key] = value

        # Call handler function
        return await route.handler(request_data)

Now our framework can automatically convert parameters to the types specified by the function annotations!

Step 4: Implement Request Body Parsing (POST Support)

Next, we'll add support for POST request bodies, enabling JSON data parsing:

# Add request body parsing to Router
async def handle(self, scope: Dict[str, Any], receive: Callable) -> Dict[str, Any]:
    # ... previous code ...

    # If it's a POST request, parse the request body
    request_body = {}
    if method == "POST":
        # Get request body from receive
        message = await receive()
        if message["type"] == "http.request" and "body" in message:
            try:
                request_body = json.loads(message["body"].decode())
            except json.JSONDecodeError:
                return {
                    "status": 400,
                    "body": {"detail": "Invalid JSON"}
                }

    # Merge all parameters
    raw_data = {** params, **query_params,** request_body}
    # ... type conversion and handler function call ...

Step 5: Build a Complete Example Application

Now we can use our framework just like FastAPI:

# main.py
from asgi_server import ASGIServer
import asyncio

# Create server instance (includes router)
app = ASGIServer()
router = app.router

# Define routes
@router.get("/")
async def root():
    return {"message": "Hello, World!"}

@router.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

@router.post("/items/")
async def create_item(name: str, price: float):
    return {"item": {"name": name, "price": price, "id": 42}}

# Run the application
if __name__ == "__main__":
    asyncio.run(app.run())

Test this application:

Visit

http://127.0.0.1:8000

→ Get welcome message
Visit http://127.0.0.1:8000/items/42?q=test → Get response with parameters
Send a POST request to http://127.0.0.1:8000/items/ with {"name": "Apple", "price": 1.99} → Get the created item

Differences from FastAPI and Optimization Directions

Our simplified version implements FastAPI's core functionality, but the real FastAPI has many advanced features:

Dependency injection system: FastAPI's dependency injection is very powerful, supporting nested dependencies, global dependencies, etc.
Automatic documentation: FastAPI can automatically generate Swagger and ReDoc documentation
More data type support: Including Pydantic model validation, form data, file uploads, etc.
Middleware system: More complete middleware support
WebSocket support: Full implementation of ASGI's WebSocket specification
Asynchronous database tools: Deep integration with tools like SQLAlchemy

Summary: What Have We Learned?

Through this hands-on practice, we've understood:

The basic working principles of the ASGI protocol: the three elements of scope, receive, and send
The core of the routing system: path matching, parameter parsing, and handler function mapping
How type conversion is implemented: using function annotations for automatic conversion
The request handling process: the complete lifecycle from receiving a request to returning a response

This knowledge applies not only to FastAPI but also to all ASGI frameworks (like Starlette, Quart, etc.). When you encounter problems using these frameworks, recalling the simplified version we built today will help resolve many confusions.

Finally, remember: the best way to learn is through hands-on practice. Try extending our simplified framework—like adding dependency injection or more complete error handling. This will take your understanding of web frameworks to the next level!