Chio/Docs

Python HTTP Frameworks

Two Python examples that govern the same GET /hello / POST /echo contract through a local Chio sidecar. FastAPI uses an ASGI middleware; Django uses the synchronous middleware protocol.

What it shows

  • chio-asgi as a reusable ASGI middleware plus chio-fastapi for framework-native receipt access.
  • chio-django (ChioDjangoMiddleware) for standard Django request/response middleware.
  • Both call out to a real Chio sidecar over /chio/evaluate; the smoke flow lists persisted receipts from the sidecar SQLite store.
  • Request bodies remain readable by the application after the middleware hashes them.

SDK shape

The Python SDKs do not embed the kernel. They speak HTTP to a local chio api protect sidecar. That keeps the Python footprint small and avoids any FFI build step. See HTTP Framework Middleware for the deeper pattern.

Use this when

Use this when: your service runs on FastAPI/Starlette/ASGI or Django/WSGI and you want the receipt id available inside the request scope (request.state.chio_receipt for FastAPI; request.chio_receipt for Django). Don't use this if you want the zero-code reverse-proxy shape: read Protect an API and run chio api protect in front of an unmodified Python service. For Node see Node HTTP Frameworks; for Spring or ASP.NET see JVM and .NET.

Prerequisites

  • Python 3.11+ and uv.
  • The chio CLI on PATH. The smoke flows stand up chio trust serve and chio api protect for you.

Run them

bash
cd examples/hello-fastapi    # or hello-django
./run.sh

# In another shell, run the full sidecar + trust + deny + allow smoke
./smoke.sh

Default ports:

ExampleEnv varDefault
hello-fastapiHELLO_FASTAPI_PORT8011
hello-djangoHELLO_DJANGO_PORT8016

FastAPI (ASGI)

FastAPI runs under an ASGI server (uvicorn in this example). The middleware is registered as part of app construction; FastAPI applies it before route handlers run.

Before

app.py
from fastapi import FastAPI
from pydantic import BaseModel


class EchoRequest(BaseModel):
    message: str
    count: int = 1


app = FastAPI(title="hello-fastapi", version="0.1.0")


@app.get("/hello")
async def hello() -> dict[str, str]:
    return {"message": "hello from fastapi"}


@app.post("/echo")
async def echo(payload: EchoRequest) -> dict[str, object]:
    return {"message": payload.message, "count": payload.count}

After

examples/hello-fastapi/app.py
from __future__ import annotations

import os

from fastapi import FastAPI
from pydantic import BaseModel

from chio_asgi import ChioASGIMiddleware
from chio_asgi.config import ChioASGIConfig


class EchoRequest(BaseModel):
    message: str
    count: int = 1


def create_app() -> FastAPI:
    app = FastAPI(
        title="hello-fastapi",
        version="0.1.0",
        docs_url=None,
        redoc_url=None,
        openapi_url=None,
    )

    app.add_middleware(
        ChioASGIMiddleware,
        config=ChioASGIConfig(
            sidecar_url=os.environ.get("CHIO_SIDECAR_URL", "http://127.0.0.1:9090"),
            exclude_paths=frozenset({"/healthz"}),
        ),
    )

    @app.get("/healthz")
    async def healthz() -> dict[str, str]:
        return {"status": "ok"}

    @app.get("/hello")
    async def hello() -> dict[str, str]:
        return {"message": "hello from fastapi"}

    @app.post("/echo")
    async def echo(payload: EchoRequest) -> dict[str, object]:
        return {"message": payload.message, "count": payload.count}

    return app


app = create_app()

Notes: the ASGI middleware reads the request stream once, hashes the body, then replays it for FastAPI's body parser. Pydantic models receive the same bytes the sidecar evaluated. The receipt id arrives as a response header.

bash
./run.sh
# starts uvicorn on 127.0.0.1:8011 against app:app

Django (WSGI)

Django uses a synchronous middleware protocol. Registration is one string in MIDDLEWARE plus a few CHIO_* settings. The middleware decorates the request with request.chio_receipt.

Settings

examples/hello-django/hello_project/settings.py
MIDDLEWARE = [
    "chio_django.ChioDjangoMiddleware",
]

CHIO_SIDECAR_URL = os.environ.get("CHIO_SIDECAR_URL", "http://127.0.0.1:9090")
CHIO_FAIL_OPEN = False
CHIO_EXCLUDE_PATHS = ["/healthz"]
CHIO_EXCLUDE_METHODS = ["OPTIONS"]
CHIO_RECEIPT_HEADER = "X-Chio-Receipt"

CHIO_FAIL_OPEN = False: when the sidecar is unreachable, requests are denied rather than waved through.

URLs

examples/hello-django/hello_project/urls.py
from django.urls import path

from hello_app import views

urlpatterns = [
    path("healthz", views.healthz),
    path("hello", views.hello),
    path("echo", views.echo),
]

Views

examples/hello-django/hello_app/views.py
from __future__ import annotations

import json

from django.http import HttpRequest, JsonResponse


def _receipt_id(request: HttpRequest) -> str | None:
    receipt = getattr(request, "chio_receipt", None)
    if isinstance(receipt, dict):
        return receipt.get("id")
    return None


def healthz(_request: HttpRequest) -> JsonResponse:
    return JsonResponse({"status": "ok"})


def hello(request: HttpRequest) -> JsonResponse:
    return JsonResponse({
        "message": "hello from django",
        "receipt_id": _receipt_id(request),
    })


def echo(request: HttpRequest) -> JsonResponse:
    try:
        payload = json.loads(request.body.decode("utf-8") or "{}")
    except json.JSONDecodeError as exc:
        return JsonResponse({"error": str(exc)}, status=400)

    return JsonResponse({
        "message": payload.get("message"),
        "count": payload.get("count", 1),
        "receipt_id": _receipt_id(request),
        "body_cached": bool(request.body),
    })

Note: request.body is still readable inside the view even though the middleware already hashed it. The middleware caches the bytes on the request so the view does not see an exhausted stream.


Where the receipt id lands

Both middlewares attach the kernel verdict to the request and add a response header for the client.

chio-fastapi/dependencies.py:91
async def get_chio_receipt(request: Request) -> HttpReceipt | None:
    return getattr(request.state, "chio_receipt", None)

FastAPI handlers reach for request.state.chio_receipt directly, or use Depends(get_chio_receipt). Django handlers read getattr(request, "chio_receipt", None). Both responses carry x-chio-receipt (FastAPI) or X-Chio-Receipt (Django, configured via CHIO_RECEIPT_HEADER).


Smoke assertions

Each smoke.sh spins up a trust service, the app, and the sidecar; runs three curls; and asserts the response body for each.

examples/hello-fastapi/smoke.sh
# 1. GET /hello (allow)
body = json.loads(...)
assert body["message"] == "hello from fastapi", body

# 2. POST /echo without capability (deny)
body = json.loads(...)
assert body["status"] == 403, body
assert body["message"], body

# 3. POST /echo with X-Chio-Capability (allow)
body = json.loads(...)
assert body["message"] == "hello", body
assert body["count"] == 2, body
examples/hello-django/smoke.sh
# Django adds receipt-id assertions on the body
assert body["message"] == "hello from django", body
assert body["receipt_id"], body                 # /hello

assert body["error"]["code"] == "CHIO_GUARD_DENIED", body
grep -q " 403 " "${ARTIFACT_ROOT}/deny.headers"

assert body["body_cached"] is True, body        # /echo allow path

Inspect after

Artifacts land under .artifacts/<UTC-timestamp>/. After a smoke run:

bash
cd .artifacts/$(ls -t .artifacts | head -1)

# Receipt id from the response headers
grep -i x-chio-receipt hello.headers
# expect: x-chio-receipt: 01J...

# Persisted receipts (one per request)
wc -l receipts.ndjson         # expect: 3
jq -r '.id, .verdict.outcome' receipts.ndjson

# Verdict outcomes (allow/deny)
jq -r '.verdict.outcome' receipts.ndjson | sort | uniq -c
# expect: 2 allow, 1 deny

# SQLite store directly
sqlite3 state/sidecar-receipts.sqlite3 \
  "select id, substr(verdict_outcome,1,8) from receipts order by created_at desc limit 5;"

ASGI vs WSGI

AspectFastAPI / ASGIDjango / WSGI
Middleware protocolASGI scope/receive/sendSync request/response callable
SDK packagechio-asgi (+ chio-fastapi)chio-django
Receipt accessResponse header (and FastAPI helpers)request.chio_receipt + header
Body replayWraps receive to replay bytesCaches request.body
AsyncNativeSync; use ASGI Django for async views

Next