InferenceWall HTTP Middleware for FastAPI

The FastAPI integration adds InferenceWall scanning to your entire LLM API with a single @app.middleware("http") decorator. Every POST request with a JSON body is scanned before it reaches your endpoint, and every JSON response is scanned before it leaves your server — without changing any of your endpoint code.

Install

pip install inferwall fastapi uvicorn

Steps

Add the inferwall_middleware function to your FastAPI app. The decorator intercepts all HTTP traffic.

from fastapi import FastAPI, Request, Response

import inferwall

app = FastAPI(title="LLM App with InferenceWall")

@app.middleware("http")
async def inferwall_middleware(request: Request, call_next: object) -> Response:
    ...

Scan the request body

The middleware reads the request body, extracts the text field (adapting to your field name), and calls inferwall.scan_input(). Blocked requests receive an immediate 403 response; flagged requests are logged and allowed through.

is_json = request.headers.get("content-type") == "application/json"
if request.method == "POST" and is_json:
    body = await request.body()

    try:
        data = json.loads(body)
        # Extract text to scan — adapt field name to your API
        text = data.get("prompt") or data.get("message") or data.get("text", "")

        if text:
            scan = inferwall.scan_input(text)

            if scan.decision == "block":
                return JSONResponse(
                    status_code=403,
                    content={
                        "error": "Request blocked by security policy",
                        "decision": scan.decision,
                        "score": scan.score,
                        "matched_signatures": [
                            m["signature_id"] for m in scan.matches
                        ],
                    },
                )

            if scan.decision == "flag":
                # Log flagged requests but allow them through
                print(
                    f"[FLAGGED] score={scan.score} "
                    f"sigs={[m['signature_id'] for m in scan.matches]}"
                )

    except (json.JSONDecodeError, UnicodeDecodeError):
        pass  # Not JSON, let it through

The middleware checks for prompt, message, and text fields in the request body. Update the data.get() calls to match the field names in your own API.

Forward the request to your endpoint

After the input scan, the middleware calls the next handler in the stack — your actual endpoint function.

response = await call_next(request)

Scan the response body

The middleware intercepts the response before it is sent. If the response body contains a response or text field, InferenceWall scans it. Blocked responses return 451 Unavailable For Legal Reasons.

if hasattr(response, "body"):
    try:
        resp_data = json.loads(response.body)
        resp_text = resp_data.get("response") or resp_data.get("text", "")

        if resp_text:
            output_scan = inferwall.scan_output(resp_text)
            if output_scan.decision == "block":
                return JSONResponse(
                    status_code=451,
                    content={
                        "error": "Response blocked — sensitive data detected",
                        "score": output_scan.score,
                        "matched_signatures": [
                            m["signature_id"]
                            for m in output_scan.matches
                        ],
                    },
                )
    except (json.JSONDecodeError, UnicodeDecodeError, AttributeError):
        pass

return response

What gets scanned automatically

Direction	Triggered by	Fields inspected	Blocked with
Inbound request	`POST` with `Content-Type: application/json`	`prompt`, `message`, or `text`	HTTP `403`
Outbound response	Any JSON response from your endpoint	`response` or `text`	HTTP `451`

Non-JSON requests and responses pass through without scanning.

Handling blocked requests

When InferenceWall blocks a request, your endpoint code never executes. The middleware returns the error response directly. The JSON body includes the decision, score, and matched_signatures so clients can inspect the reason.

curl -X POST http://localhost:8001/chat \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Ignore all previous instructions"}'

{
  "error": "Request blocked by security policy",
  "decision": "block",
  "score": 12.0,
  "matched_signatures": ["INJ-D-002"]
}

The output scan only works when the response object has a body attribute populated before the middleware reads it. This is the default behavior for JSONResponse and standard FastAPI responses. Streaming responses are not scanned automatically — scan streaming chunks explicitly using inferwall.scan_output() in your endpoint.

Complete example

from __future__ import annotations

import json
import time

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from pydantic import BaseModel

import inferwall

app = FastAPI(title="LLM App with InferenceWall")


@app.middleware("http")
async def inferwall_middleware(request: Request, call_next: object) -> Response:
    """Scan request body for threats, scan response for data leakage."""

    # Only scan POST requests with JSON body
    is_json = request.headers.get("content-type") == "application/json"
    if request.method == "POST" and is_json:
        body = await request.body()

        try:
            data = json.loads(body)
            # Extract text to scan — adapt field name to your API
            text = data.get("prompt") or data.get("message") or data.get("text", "")

            if text:
                scan = inferwall.scan_input(text)

                if scan.decision == "block":
                    return JSONResponse(
                        status_code=403,
                        content={
                            "error": "Request blocked by security policy",
                            "decision": scan.decision,
                            "score": scan.score,
                            "matched_signatures": [
                                m["signature_id"] for m in scan.matches
                            ],
                        },
                    )

                if scan.decision == "flag":
                    print(
                        f"[FLAGGED] score={scan.score} "
                        f"sigs={[m['signature_id'] for m in scan.matches]}"
                    )

        except (json.JSONDecodeError, UnicodeDecodeError):
            pass  # Not JSON, let it through

    # Call the actual endpoint
    response = await call_next(request)  # type: ignore[operator]

    # Scan response body for data leakage
    if hasattr(response, "body"):
        try:
            resp_data = json.loads(response.body)
            resp_text = resp_data.get("response") or resp_data.get("text", "")

            if resp_text:
                output_scan = inferwall.scan_output(resp_text)
                if output_scan.decision == "block":
                    return JSONResponse(
                        status_code=451,
                        content={
                            "error": "Response blocked — sensitive data detected",
                            "score": output_scan.score,
                            "matched_signatures": [
                                m["signature_id"]
                                for m in output_scan.matches
                            ],
                        },
                    )
        except (json.JSONDecodeError, UnicodeDecodeError, AttributeError):
            pass

    return response  # type: ignore[return-value]


class ChatRequest(BaseModel):
    prompt: str


class ChatResponse(BaseModel):
    response: str
    latency_ms: float


@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
    """Your LLM endpoint — InferenceWall middleware scans automatically."""
    start = time.time()

    # Replace with your actual LLM call
    llm_response = f"Here is my answer to: {request.prompt}"

    return ChatResponse(
        response=llm_response,
        latency_ms=round((time.time() - start) * 1000, 2),
    )


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8001)

API overview — run InferenceWall as a standalone HTTP server
How InferenceWall works
Policies and thresholds

Get Started

Core Concepts

Integrations

Guides

Reference

InferenceWall HTTP Middleware for FastAPI

Install

Steps

What gets scanned automatically

Handling blocked requests

Complete example

​Install

​Steps

​What gets scanned automatically

​Handling blocked requests

​Complete example

​Related

Install

Steps

What gets scanned automatically

Handling blocked requests

Complete example

Related