FastAPI

How Beam reads a FastAPI app

There's no in-process SDK for this stack, so the app writes what Beam needs to its logs and beam monitor reads them on the same server or container. Only redacted findings leave it. Two kinds of line:

One JSON line per request, with method, path (with the query string), status, ip, user_agent, host and duration_ms. The status is what lets Beam tell an attack your app refused from one it answered.
Security events as plain lines, worded like this (Beam matches the wording and reads user= and ip=):

login failed user=42 ip=203.0.113.7
login succeeded user=42 ip=203.0.113.7
granted admin role to user=99 by user=12
API key created user=12
{"event":"llm_usage","user":"42","model":"gpt-5","total_tokens":1834}

Use an opaque user id, never an email. New outbound hosts are only reported by the Node.js SDK; for this stack, restrict outbound traffic at the network instead.

Uvicorn's own access log ("GET /x HTTP/1.1" 200 OK) is read too, but without its status, so use the JSON line below.

Before you start

Do steps 1 and 2 of the general cookbook first: App monitoring is on, the app is added, and its beam_app_ token is in your secret store as BEAM_APP_TOKEN. Everything below needs @agent-beam/beam 0.2.32 or newer.

Log each request with its status

# beam_logging.py
import json, logging, logging.handlers, time
from fastapi import FastAPI, Request

def _logger(name, path):
    handler = logging.handlers.WatchedFileHandler(path)
    handler.setFormatter(logging.Formatter("%(message)s"))
    logger = logging.getLogger(name)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False
    return logger

requests_log = _logger("beam.requests", "/var/log/app/requests.log")
security = _logger("beam.security", "/var/log/app/security.log")

def client_ip(request: Request) -> str:
    forwarded = request.headers.get("x-forwarded-for")
    return forwarded.split(",")[0].strip() if forwarded else (request.client.host if request.client else "")

def init_beam(app: FastAPI):
    @app.middleware("http")
    async def log_request(request: Request, call_next):
        started = time.monotonic()
        response = await call_next(request)
        query = f"?{request.url.query}" if request.url.query else ""
        requests_log.info(json.dumps({
            "method": request.method,
            "path": request.url.path + query,
            "status": response.status_code,
            "ip": client_ip(request),
            "user_agent": request.headers.get("user-agent", ""),
            "host": request.headers.get("host", ""),
            "duration_ms": round((time.monotonic() - started) * 1000),
        }))
        return response

Call init_beam(app) right after app = FastAPI().

Log sign-ins, admins, keys and LLM spend

from beam_logging import security, client_ip

@app.post("/token")
async def login(request: Request, form: OAuth2PasswordRequestForm = Depends()):
    user = await verify_password(form.username, form.password)
    if not user:
        security.warning("login failed user=%s ip=%s", await user_id_for(form.username), client_ip(request))
        raise HTTPException(status_code=401)
    security.info("login succeeded user=%s ip=%s", user.id, client_ip(request))
    return issue_token(user)

# After a model call
security.info(json.dumps({"event": "llm_usage", "user": str(user.id), "model": model, "total_tokens": usage.total_tokens}))

Run `beam monitor` next to the app

npm install -g @agent-beam/beam
beam monitor --stack fastapi --file /var/log/app/requests.log --file /var/log/app/security.log --app /srv/app

--stack fastapi turns on this framework's rules and rolls up attacks on frameworks you don't run; --app also checks the app's lockfiles daily (and would detect the stack on its own). Keep it running with systemd:

# /etc/systemd/system/beam-monitor.service
[Unit]
Description=AgentBeam App monitoring
After=network-online.target

[Service]
EnvironmentFile=/etc/beam/app.env
ExecStart=/usr/bin/env beam monitor --stack fastapi --file /var/log/app/requests.log --file /var/log/app/security.log --app /srv/app
Restart=always
User=beam

[Install]
WantedBy=multi-user.target

sudo install -m 600 /dev/null /etc/beam/app.env && echo "BEAM_APP_TOKEN=beam_app_..." | sudo tee /etc/beam/app.env >/dev/null
sudo systemctl daemon-reload && sudo systemctl enable --now beam-monitor

In Docker or Kubernetes, run the same command in a sidecar that mounts the log directory read-only, with BEAM_APP_TOKEN from a secret.

Prove it works

beam monitor status   # app, level, stack, usage
beam monitor test     # a test finding through your alert workflows
curl -s "https://your-app.example.com/?id=1%20UNION%20SELECT%201" -o /dev/null

Within a minute, the SQL injection test shows on App monitoring: under Needs action if the app answered 2xx, or counted under "needed nothing from you" if it refused it. Then sign in to a test account with a wrong password ten times and the right one once: Sign-in succeeded after 10 failed attempts appears as critical.

Check dependencies on every pull request

# .github/workflows/beam-cve.yml
name: Dependency check
on: [pull_request]
jobs:
  cve:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: npx -y @agent-beam/beam cve check . --fail-on high
        env:
          BEAM_APP_TOKEN: ${{ secrets.BEAM_APP_TOKEN }}

Pin your Python dependencies (uv.lock, poetry.lock or requirements.txt with ==).

What Beam watches for on this stack

Besides every built-in rule and the attack-in-progress signals:

Rule	Means	Severity
GraphQL introspection	A `__schema` query in a GET, if you serve Strawberry or Ariadne	Medium
Attacks on AI features	Prompt injection and key extraction in query strings and logged prompts	High

Attacks on other frameworks (WordPress probes, Laravel or Spring exploits, and so on) are rolled up as Not your stack when there's no status to say otherwise, and never alert.

Close what Beam would catch

Beam raises these when they're found; better that there's nothing to find:

Decide whether /docs and /openapi.json should be public; set docs_url=None, openapi_url=None if not.
Run uvicorn with --proxy-headers --forwarded-allow-ips set to your proxy only.

Coverage checklist

Feature	Done when
Request log with status	Findings show Answered, Refused or Errored
Stack declared	`beam monitor` prints the stack at start
Sign-in events	`login failed` and `login succeeded` lines are written
Admin and key events	Role grants and new keys are logged
LLM spend	An `llm_usage` line follows each model call
Dependency checks	The CI job runs on pull requests
Alerts	Critical pages someone, high reaches a channel (step 8)
Routine	Someone reads Needs action daily and Attack analytics weekly