> ## Documentation Index
> Fetch the complete documentation index at: https://docs.inferwall.com/llms.txt
> Use this file to discover all available pages before exploring further.

# InferenceWall HTTP Middleware for FastAPI

> Automatically scan every incoming HTTP request body and every outgoing response body with InferenceWall by registering a single FastAPI middleware function.

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

```bash theme={null}
pip install inferwall fastapi uvicorn
```

## Steps

<Steps>
  <Step title="Register the middleware">
    Add the `inferwall_middleware` function to your FastAPI app. The decorator intercepts all HTTP traffic.

    ```python theme={null}
    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:
        ...
    ```
  </Step>

  <Step title="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.

    ```python theme={null}
    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
    ```

    <Note>
      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.
    </Note>
  </Step>

  <Step title="Forward the request to your endpoint">
    After the input scan, the middleware calls the next handler in the stack — your actual endpoint function.

    ```python theme={null}
    response = await call_next(request)
    ```
  </Step>

  <Step title="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`.

    ```python theme={null}
    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
    ```
  </Step>
</Steps>

## 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.

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

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

<Warning>
  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.
</Warning>

## Complete example

```python theme={null}
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)
```

## Related

* [API overview](/api/overview) — run InferenceWall as a standalone HTTP server
* [How InferenceWall works](/concepts/how-it-works)
* [Policies and thresholds](/concepts/policies)