# Sample code

> Complete Node.js and Python programs for the Fileverity API: check an image, check a large PDF, detect AI-written text and verify webhook signatures.

Source: https://fileverity.com/en/docs/sample-code

This page has four complete programs for the Fileverity API, each in Node.js 18 or later and in Python 3.9 or later: check an image and wait for the result, check a large PDF, detect AI-written text, and receive webhooks with signature verification. Copy one, set `FILEVERITY_API_KEY` and run it.

The programs call the HTTPS API directly, with the `fetch` built into Node.js and the `requests` package in Python, so all you need is an HTTP client. The calls, fields and statuses are the ones described in [Files](https://fileverity.com/en/docs/files.md), [Analyses](https://fileverity.com/en/docs/analyses.md), [Text detections](https://fileverity.com/en/docs/text-detections.md) and [Webhooks](https://fileverity.com/en/docs/webhooks.md).

| Program | What it does | Calls |
| --- | --- | --- |
| [Check an image](https://fileverity.com/en/docs/sample-code.md#image) | Uploads an image, waits for the result and prints the summary and the report URL. | `POST /files`, `POST /analyses`, `GET /analyses/{id}` |
| [Check a large PDF](https://fileverity.com/en/docs/sample-code.md#pdf) | Uploads a document with a pre-signed URL, polls until the check finishes and saves the PDF report. | `POST /files`, `GET /files/{id}`, `POST /analyses`, `GET /analyses/{id}`, `GET /analyses/{id}/report.pdf` |
| [Detect AI-written text](https://fileverity.com/en/docs/sample-code.md#text) | Sends a text file and prints the verdict, the score and each AI-written passage. | `POST /text/detections` |
| [Receive webhooks](https://fileverity.com/en/docs/sample-code.md#webhooks) | Verifies the signature of each delivery, ignores duplicates and handles the events. | Your endpoint, called with `analysis.completed`, `analysis.failed` and `balance.low` |

## Before you run

- Create a **Test** key under [API keys](https://fileverity.com/console/keys) in the Console. A test key returns fixed sample results and is never charged, so every program runs at no cost. To check real files, create a **Live** key. The programs do not change.
- Node.js: save each program as an `.mjs` file, which lets it use `import` and top-level `await`, and run it with `node`. Nothing to install.
- Python: run `pip install requests`. The webhook receiver also needs `pip install flask`.

**Set the key: Shell**

```bash
export FILEVERITY_API_KEY=fv_test_...
```

**Set the key: PowerShell**

```powershell
$env:FILEVERITY_API_KEY = "fv_test_..."
```

## How the programs handle keys, errors and retries

The three programs that call the API share the same `request` and `api` functions. Copy them into your own code as they are.

| Concern | What the programs do |
| --- | --- |
| Key | Read from `FILEVERITY_API_KEY` and never written into the code. Call the API from your server, not from browser code. |
| Idempotency | Every POST sends an `Idempotency-Key`, a new UUID per request, and reuses it on every retry, so a retry never runs or bills a check twice. To make running the whole script again safe, derive the key from your own record ID instead. See [Idempotency](https://fileverity.com/en/docs/idempotency.md). |
| Errors | A response that is not `2xx` carries `error.type`, `error.code`, `error.message`, `error.param` and `error.request_id`. The programs stop with `code: message (request request_id)`. Branch on `code` and quote the request ID to support. See [Errors](https://fileverity.com/en/docs/errors.md). |
| Retries | A network failure, a `5xx` status (`internal_error`, `analysis_failed`, `service_unavailable`), `rate_limited` (429) and `idempotency_in_progress` (409) are retried, up to 5 attempts. The wait is the `retry-after` header when the response has one, otherwise 2, 4, 8 and 16 seconds. Every other `4xx` stops at once. `pending_files_limit` is also a 429 but has no `retry-after`, so it is not retried. |
| Rate limit | When a response reports `x-ratelimit-remaining-requests: 0`, the program waits `x-ratelimit-reset-requests` seconds before the next call, so it does not run into a 429. See [Rate limits and headers](https://fileverity.com/en/docs/rate-limits.md#headers). |
| Timeouts | The Python programs time out after 90 seconds per request, longer than the 60 seconds that `Prefer: wait` can hold a request. |

## Check an image and wait for the result

The program uploads the image as multipart, creates an analysis with `Prefer: wait=60` so the request is held until the result is ready, and prints the summary and the report URL. If the wait ends first, the API answers `202` with `status: processing`, and the program polls `GET /analyses/{id}` every 3 seconds. See [Wait, poll or use webhooks](https://fileverity.com/en/docs/analyses.md#wait).

Save it as `check-image.mjs` or `check_image.py` and run `node check-image.mjs photo.jpg` or `python check_image.py photo.jpg`. Multipart takes images up to 4 MB. For a larger image, up to 10 MB, use the pre-signed upload in the next program with the content type of the image, for example `image/jpeg`.

**Check an image: JavaScript**

```javascript
import { randomUUID } from "node:crypto";
import { readFile } from "node:fs/promises";
import { basename } from "node:path";

const base = "https://api.fileverity.com/v1";
const key = process.env.FILEVERITY_API_KEY;
const path = process.argv[2];
if (!key || !path) throw new Error("Usage: FILEVERITY_API_KEY=... node check-image.mjs photo.jpg");

const retryable = ["rate_limited", "idempotency_in_progress"];
const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000));

async function request(url, init) {
  for (let attempt = 1; ; attempt++) {
    let response = null;
    let problem;
    try {
      response = await fetch(url, init);
      if (response.ok) {
        const remaining = response.headers.get("x-ratelimit-remaining-requests");
        const reset = Number(response.headers.get("x-ratelimit-reset-requests"));
        if (remaining === "0") await sleep(reset);
        return response;
      }
      problem = (await response.json().catch(() => ({}))).error ?? {};
    } catch (error) {
      problem = { code: "network_error", message: error.message };
    }
    const status = response?.status ?? 0;
    const retry = status === 0 || status >= 500 || retryable.includes(problem.code);
    if (!retry || attempt === 5) {
      const { code = status, message = "Request failed", request_id: id = "unknown" } = problem;
      throw new Error(`${code}: ${message} (request ${id})`);
    }
    await sleep(Number(response?.headers.get("retry-after")) || 2 ** attempt);
  }
}

async function api(method, route, { headers, body } = {}) {
  const auth = { Authorization: `Bearer ${key}` };
  const response = await request(base + route, { method, headers: { ...auth, ...headers }, body });
  return response.json();
}

const form = new FormData();
form.append("file", new Blob([await readFile(path)]), basename(path));
const file = await api("POST", "/files", {
  headers: { "Idempotency-Key": randomUUID() },
  body: form,
});

let analysis = await api("POST", "/analyses", {
  headers: {
    "Content-Type": "application/json",
    "Idempotency-Key": randomUUID(),
    Prefer: "wait=60",
  },
  body: JSON.stringify({ model: "fvimg-2", file: file.id }),
});
while (analysis.status === "processing") {
  await sleep(3);
  analysis = await api("GET", `/analyses/${analysis.id}`);
}

if (analysis.status !== "succeeded") {
  throw new Error(`The check ${analysis.status}: ${analysis.error?.message}`);
}
const { level, action, assessment, ai, edit } = analysis.summary;
console.log(`${analysis.id} ${analysis.status}`);
console.log(`level ${level}, action ${action}, assessment ${assessment}`);
console.log(`AI generation: ${ai}, editing: ${edit}`);
console.log(`Report: ${analysis.report_url}`);
```

**Check an image: Python**

```python
import os
import sys
import time
import uuid

import requests

BASE = "https://api.fileverity.com/v1"
KEY = os.environ.get("FILEVERITY_API_KEY")
if not KEY or len(sys.argv) < 2:
    sys.exit("Usage: FILEVERITY_API_KEY=... python check_image.py photo.jpg")

RETRYABLE = ("rate_limited", "idempotency_in_progress")


def request(method, url, **kwargs):
    for attempt in range(1, 6):
        try:
            response = requests.request(method, url, timeout=90, **kwargs)
        except requests.RequestException as error:
            response, problem = None, {"code": "network_error", "message": str(error)}
        else:
            if response.ok:
                if response.headers.get("x-ratelimit-remaining-requests") == "0":
                    time.sleep(int(response.headers["x-ratelimit-reset-requests"]))
                return response
            try:
                problem = response.json()["error"]
            except (ValueError, KeyError, TypeError):
                problem = {}
        status = response.status_code if response is not None else 0
        retry = status == 0 or status >= 500 or problem.get("code") in RETRYABLE
        if not retry or attempt == 5:
            code = problem.get("code", status)
            message = problem.get("message", "Request failed")
            request_id = problem.get("request_id", "unknown")
            raise RuntimeError(f"{code}: {message} (request {request_id})")
        wait = response.headers.get("retry-after") if response is not None else None
        time.sleep(int(wait or 2**attempt))


def api(method, route, headers=None, **kwargs):
    headers = {"Authorization": f"Bearer {KEY}", **(headers or {})}
    return request(method, BASE + route, headers=headers, **kwargs).json()


with open(sys.argv[1], "rb") as source:
    data = source.read()
upload = {"file": (os.path.basename(sys.argv[1]), data)}
file = api("POST", "/files", headers={"Idempotency-Key": str(uuid.uuid4())}, files=upload)

analysis = api(
    "POST",
    "/analyses",
    headers={"Idempotency-Key": str(uuid.uuid4()), "Prefer": "wait=60"},
    json={"model": "fvimg-2", "file": file["id"]},
)
while analysis["status"] == "processing":
    time.sleep(3)
    analysis = api("GET", f"/analyses/{analysis['id']}")

if analysis["status"] != "succeeded":
    problem = analysis["error"] or {}
    sys.exit(f"The check {analysis['status']}: {problem.get('message')}")
summary = analysis["summary"]
print(f"{analysis['id']} {analysis['status']}")
print(f"level {summary['level']}, action {summary['action']}, assessment {summary['assessment']}")
print(f"AI generation: {summary['ai']}, editing: {summary['edit']}")
print(f"Report: {analysis['report_url']}")
```

**Output of the image check**

```text
an_8Hk2pQ4vN9sL1xR7cW3mY6zB succeeded
level L2, action accept, assessment authentic_untouched
AI generation: very_low, editing: none
Report: https://fileverity.com/verify/FV-7K2M-9QXD-4T8R
```

`summary` is the stable, machine-readable verdict of an image check: `level`, `action`, `assessment`, `ai` and `edit`, described in [Image summary](https://fileverity.com/en/docs/analyses.md#summary-image). `report_url` is the public verification page of the report. With a test key the result is a fixed sample, and its report URL has no page behind it.

## Check a large PDF

Documents up to 50 MB are uploaded in three steps: ask for an upload URL with `POST /files` and a JSON body, send the bytes to that URL with `PUT`, and wait until the file shows `uploaded`. The program then creates the analysis and polls every 5 seconds, because documents can be queued and a check can take a while. It prints the progress, the summary and the report URL, and saves the PDF report next to the script.

Save it as `check-pdf.mjs` or `check_pdf.py` and run `node check-pdf.mjs contract.pdf` or `python check_pdf.py contract.pdf`. For another document type, change `content_type` to one of the [accepted content types](https://fileverity.com/en/docs/files.md#content-types). For an encrypted document, add `options: { password }` to the analysis request.

**Check a large PDF: JavaScript**

```javascript
import { randomUUID } from "node:crypto";
import { readFile, writeFile } from "node:fs/promises";
import { basename } from "node:path";

const base = "https://api.fileverity.com/v1";
const key = process.env.FILEVERITY_API_KEY;
const path = process.argv[2];
if (!key || !path) throw new Error("Usage: FILEVERITY_API_KEY=... node check-pdf.mjs contract.pdf");

const retryable = ["rate_limited", "idempotency_in_progress"];
const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000));

async function request(url, init) {
  for (let attempt = 1; ; attempt++) {
    let response = null;
    let problem;
    try {
      response = await fetch(url, init);
      if (response.ok) {
        const remaining = response.headers.get("x-ratelimit-remaining-requests");
        const reset = Number(response.headers.get("x-ratelimit-reset-requests"));
        if (remaining === "0") await sleep(reset);
        return response;
      }
      problem = (await response.json().catch(() => ({}))).error ?? {};
    } catch (error) {
      problem = { code: "network_error", message: error.message };
    }
    const status = response?.status ?? 0;
    const retry = status === 0 || status >= 500 || retryable.includes(problem.code);
    if (!retry || attempt === 5) {
      const { code = status, message = "Request failed", request_id: id = "unknown" } = problem;
      throw new Error(`${code}: ${message} (request ${id})`);
    }
    await sleep(Number(response?.headers.get("retry-after")) || 2 ** attempt);
  }
}

async function api(method, route, { headers, body } = {}) {
  const auth = { Authorization: `Bearer ${key}` };
  const response = await request(base + route, { method, headers: { ...auth, ...headers }, body });
  return response.json();
}

const bytes = await readFile(path);
const file = await api("POST", "/files", {
  headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID() },
  body: JSON.stringify({
    filename: basename(path),
    content_type: "application/pdf",
    bytes: bytes.length,
  }),
});

if (file.status === "pending") {
  const { url, method, headers } = file.upload;
  await request(url, { method, headers, body: bytes });
  let ready = false;
  for (let attempt = 0; attempt < 15 && !ready; attempt++) {
    ready = (await api("GET", `/files/${file.id}`)).status === "uploaded";
    if (!ready) await sleep(2);
  }
  if (!ready) throw new Error("The upload was not confirmed.");
}
console.log(`${file.id} uploaded`);

let analysis = await api("POST", "/analyses", {
  headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID() },
  body: JSON.stringify({ model: "fvdoc-3", file: file.id }),
});
const deadline = Date.now() + 15 * 60 * 1000;
while (analysis.status === "processing") {
  if (Date.now() > deadline) {
    throw new Error(`Still processing. Fetch it later with GET /analyses/${analysis.id}`);
  }
  console.log(`${analysis.id} processing: ${analysis.progress?.stage ?? "starting"}`);
  await sleep(5);
  analysis = await api("GET", `/analyses/${analysis.id}`);
}

if (analysis.status !== "succeeded") {
  throw new Error(`The check ${analysis.status}: ${analysis.error?.message}`);
}
const { level, action, reason, pages, images, text, findings } = analysis.summary;
console.log(`${analysis.id} ${analysis.status}`);
console.log(`level ${level}, action ${action}, reason ${reason ?? "none"}`);
console.log(`${pages} pages, ${images} photos examined, text: ${text ?? "not evaluated"}`);
console.log(`findings: ${findings.join(", ") || "none"}`);
console.log(`Report: ${analysis.report_url}`);

if (analysis.livemode) {
  const pdf = await request(`${base}/analyses/${analysis.id}/report.pdf`, {
    headers: { Authorization: `Bearer ${key}` },
  });
  const name = `Fileverity-${analysis.report_id}.pdf`;
  await writeFile(name, Buffer.from(await pdf.arrayBuffer()));
  console.log(`Saved ${name}`);
}
```

**Check a large PDF: Python**

```python
import os
import sys
import time
import uuid

import requests

BASE = "https://api.fileverity.com/v1"
KEY = os.environ.get("FILEVERITY_API_KEY")
if not KEY or len(sys.argv) < 2:
    sys.exit("Usage: FILEVERITY_API_KEY=... python check_pdf.py contract.pdf")

RETRYABLE = ("rate_limited", "idempotency_in_progress")


def request(method, url, **kwargs):
    for attempt in range(1, 6):
        try:
            response = requests.request(method, url, timeout=90, **kwargs)
        except requests.RequestException as error:
            response, problem = None, {"code": "network_error", "message": str(error)}
        else:
            if response.ok:
                if response.headers.get("x-ratelimit-remaining-requests") == "0":
                    time.sleep(int(response.headers["x-ratelimit-reset-requests"]))
                return response
            try:
                problem = response.json()["error"]
            except (ValueError, KeyError, TypeError):
                problem = {}
        status = response.status_code if response is not None else 0
        retry = status == 0 or status >= 500 or problem.get("code") in RETRYABLE
        if not retry or attempt == 5:
            code = problem.get("code", status)
            message = problem.get("message", "Request failed")
            request_id = problem.get("request_id", "unknown")
            raise RuntimeError(f"{code}: {message} (request {request_id})")
        wait = response.headers.get("retry-after") if response is not None else None
        time.sleep(int(wait or 2**attempt))


def api(method, route, headers=None, **kwargs):
    headers = {"Authorization": f"Bearer {KEY}", **(headers or {})}
    return request(method, BASE + route, headers=headers, **kwargs).json()


with open(sys.argv[1], "rb") as source:
    data = source.read()
file = api(
    "POST",
    "/files",
    headers={"Idempotency-Key": str(uuid.uuid4())},
    json={
        "filename": os.path.basename(sys.argv[1]),
        "content_type": "application/pdf",
        "bytes": len(data),
    },
)

if file["status"] == "pending":
    request("PUT", file["upload"]["url"], headers=file["upload"]["headers"], data=data)
    for _ in range(15):
        if api("GET", f"/files/{file['id']}")["status"] == "uploaded":
            break
        time.sleep(2)
    else:
        sys.exit("The upload was not confirmed.")
print(f"{file['id']} uploaded")

analysis = api(
    "POST",
    "/analyses",
    headers={"Idempotency-Key": str(uuid.uuid4())},
    json={"model": "fvdoc-3", "file": file["id"]},
)
deadline = time.time() + 15 * 60
while analysis["status"] == "processing":
    if time.time() > deadline:
        sys.exit(f"Still processing. Fetch it later with GET /analyses/{analysis['id']}")
    stage = (analysis["progress"] or {}).get("stage", "starting")
    print(f"{analysis['id']} processing: {stage}")
    time.sleep(5)
    analysis = api("GET", f"/analyses/{analysis['id']}")

if analysis["status"] != "succeeded":
    problem = analysis["error"] or {}
    sys.exit(f"The check {analysis['status']}: {problem.get('message')}")
summary = analysis["summary"]
reason = summary["reason"] or "none"
text = summary["text"] or "not evaluated"
findings = ", ".join(summary["findings"]) or "none"
print(f"{analysis['id']} {analysis['status']}")
print(f"level {summary['level']}, action {summary['action']}, reason {reason}")
print(f"{summary['pages']} pages, {summary['images']} photos examined, text: {text}")
print(f"findings: {findings}")
print(f"Report: {analysis['report_url']}")

if analysis["livemode"]:
    url = f"{BASE}/analyses/{analysis['id']}/report.pdf"
    pdf = request("GET", url, headers={"Authorization": f"Bearer {KEY}"})
    name = f"Fileverity-{analysis['report_id']}.pdf"
    with open(name, "wb") as target:
        target.write(pdf.content)
    print(f"Saved {name}")
```

**Output of the PDF check**

```text
file_Pw3mY6zB8Hk2pQ4vN9sL1xR7 uploaded
an_Zc4vB7nP1sK8tL3wR6yQm2Xa processing: queued
an_Zc4vB7nP1sK8tL3wR6yQm2Xa processing: analyzing
an_Zc4vB7nP1sK8tL3wR6yQm2Xa processing: reporting
an_Zc4vB7nP1sK8tL3wR6yQm2Xa succeeded
level L2, action accept, reason clean
12 pages, 3 photos examined, text: human
findings: none
Report: https://fileverity.com/verify/FV-3PXH-6WQ2-8NDC
Saved Fileverity-FV-3PXH-6WQ2-8NDC.pdf
```

- The `PUT` goes straight to storage. It takes no `Authorization` header, and the headers in `upload.headers` are sent exactly as returned. See [Pre-signed upload](https://fileverity.com/en/docs/files.md#presigned).
- With a test key the file comes back `uploaded` and there is nothing to send, so the program skips the upload. The analysis returns a fixed sample at once, and there is no sample PDF, so the program saves the report only for a live key.
- Download the PDF as soon as the check finishes. The page images and text excerpts of a document are kept for 3 hours, and a PDF downloaded later leaves them out. See [PDF report](https://fileverity.com/en/docs/analyses.md#pdf).
- A check that ends with `failed` or `canceled` was not charged. The program stops with the error message of the analysis.

## Detect AI-written text

Text detection takes a single synchronous call. The program reads a text file, sends it to `POST /text/detections` and prints the verdict, the score, the share of AI-written text and every AI-written passage with its offsets. AI-written text detection supports English, French, German, Spanish and Chinese (`en`, `fr`, `de`, `es`, `zh`). The language is detected automatically; pass a code as the second argument to set it. The text must have at least 100 words and at most 100,000 characters.

Save it as `detect-text.mjs` or `detect_text.py` and run `node detect-text.mjs essay.txt` or `python detect_text.py essay.txt fr`. The [Quickstart](https://fileverity.com/en/docs/quickstart.md#text) has a command that writes a sample `essay.txt`. With a test key the response is a fixed result for human-written text, so the program prints no passage.

**Detect AI-written text: JavaScript**

```javascript
import { randomUUID } from "node:crypto";
import { readFile } from "node:fs/promises";

const base = "https://api.fileverity.com/v1";
const key = process.env.FILEVERITY_API_KEY;
const [, , path, language] = process.argv;
if (!key || !path) {
  throw new Error("Usage: FILEVERITY_API_KEY=... node detect-text.mjs essay.txt [en|fr|de|es|zh]");
}

const retryable = ["rate_limited", "idempotency_in_progress"];
const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000));

async function request(url, init) {
  for (let attempt = 1; ; attempt++) {
    let response = null;
    let problem;
    try {
      response = await fetch(url, init);
      if (response.ok) {
        const remaining = response.headers.get("x-ratelimit-remaining-requests");
        const reset = Number(response.headers.get("x-ratelimit-reset-requests"));
        if (remaining === "0") await sleep(reset);
        return response;
      }
      problem = (await response.json().catch(() => ({}))).error ?? {};
    } catch (error) {
      problem = { code: "network_error", message: error.message };
    }
    const status = response?.status ?? 0;
    const retry = status === 0 || status >= 500 || retryable.includes(problem.code);
    if (!retry || attempt === 5) {
      const { code = status, message = "Request failed", request_id: id = "unknown" } = problem;
      throw new Error(`${code}: ${message} (request ${id})`);
    }
    await sleep(Number(response?.headers.get("retry-after")) || 2 ** attempt);
  }
}

async function api(method, route, { headers, body } = {}) {
  const auth = { Authorization: `Bearer ${key}` };
  const response = await request(base + route, { method, headers: { ...auth, ...headers }, body });
  return response.json();
}

const text = await readFile(path, "utf8");
const detection = await api("POST", "/text/detections", {
  headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID() },
  body: JSON.stringify({ text, language }),
});

const { verdict, score, ai_fraction: share, usage, segments } = detection;
console.log(`verdict: ${verdict} (score ${score.toFixed(2)})`);
console.log(`AI-written share: ${Math.round(share * 100)}% of ${usage.characters} characters`);
console.log(`language: ${detection.language}, billed units: ${usage.billed_units}`);
const points = Array.from(text);
for (const { start, end, score: passage } of segments) {
  const excerpt = points.slice(start, Math.min(end, start + 80)).join("");
  console.log(`${start}-${end} score ${passage.toFixed(2)}: ${excerpt}`);
}
```

**Detect AI-written text: Python**

```python
import os
import sys
import time
import uuid

import requests

BASE = "https://api.fileverity.com/v1"
KEY = os.environ.get("FILEVERITY_API_KEY")
if not KEY or len(sys.argv) < 2:
    sys.exit("Usage: FILEVERITY_API_KEY=... python detect_text.py essay.txt [en|fr|de|es|zh]")
sys.stdout.reconfigure(encoding="utf-8")

RETRYABLE = ("rate_limited", "idempotency_in_progress")


def request(method, url, **kwargs):
    for attempt in range(1, 6):
        try:
            response = requests.request(method, url, timeout=90, **kwargs)
        except requests.RequestException as error:
            response, problem = None, {"code": "network_error", "message": str(error)}
        else:
            if response.ok:
                if response.headers.get("x-ratelimit-remaining-requests") == "0":
                    time.sleep(int(response.headers["x-ratelimit-reset-requests"]))
                return response
            try:
                problem = response.json()["error"]
            except (ValueError, KeyError, TypeError):
                problem = {}
        status = response.status_code if response is not None else 0
        retry = status == 0 or status >= 500 or problem.get("code") in RETRYABLE
        if not retry or attempt == 5:
            code = problem.get("code", status)
            message = problem.get("message", "Request failed")
            request_id = problem.get("request_id", "unknown")
            raise RuntimeError(f"{code}: {message} (request {request_id})")
        wait = response.headers.get("retry-after") if response is not None else None
        time.sleep(int(wait or 2**attempt))


def api(method, route, headers=None, **kwargs):
    headers = {"Authorization": f"Bearer {KEY}", **(headers or {})}
    return request(method, BASE + route, headers=headers, **kwargs).json()


with open(sys.argv[1], encoding="utf-8") as source:
    text = source.read()
language = sys.argv[2] if len(sys.argv) > 2 else None
detection = api(
    "POST",
    "/text/detections",
    headers={"Idempotency-Key": str(uuid.uuid4())},
    json={"text": text, "language": language},
)

usage = detection["usage"]
share = round(detection["ai_fraction"] * 100)
print(f"verdict: {detection['verdict']} (score {detection['score']:.2f})")
print(f"AI-written share: {share}% of {usage['characters']} characters")
print(f"language: {detection['language']}, billed units: {usage['billed_units']}")
for segment in detection["segments"]:
    start, end = segment["start"], segment["end"]
    excerpt = text[start : min(end, start + 80)]
    print(f"{start}-{end} score {segment['score']:.2f}: {excerpt}")
```

**Output for the Quickstart sample**

```text
verdict: mixed (score 0.97)
AI-written share: 50% of 889 characters
language: en, billed units: 1
267-495 score 0.97: In today's fast-paced logistics landscape, it is essential for businesses to pri
590-809 score 0.93: Moving forward, it is crucial to implement robust quality assurance measures and
```

`start` and `end` count Unicode code points, not bytes and not UTF-16 units. Python strings already index by code point; the Node.js program splits the text with `Array.from` first. See [Segment offsets](https://fileverity.com/en/docs/text-detections.md#offsets).

> **Note.** Text detection returns a statistical estimate, not proof of authorship. Use it as one input to a human review, and do not use it alone to make decisions that affect a person.

## Receive webhooks and verify signatures

A webhook tells your server when a check finishes, so you do not have to poll. This receiver is a minimal HTTP server: the `http` module of Node.js, and Flask for Python. It verifies the signature of every delivery, ignores a delivery it has already handled and prints the event. It follows the [Webhooks](https://fileverity.com/en/docs/webhooks.md#signature) page.

1. Make port 3000 reachable at a public HTTPS address, for example with a tunnel to your machine. Endpoint URLs must use HTTPS on the default port 443 and point to a public address.
2. In the Console, open **Webhooks**, add an endpoint with the URL `https://your-host/webhooks/fileverity` and choose its events. Copy the signing secret, which starts with `whsec_`. It is shown once.
3. Save the program as `webhook-receiver.mjs` or `webhook_receiver.py` and start it with the secret: `FILEVERITY_WEBHOOK_SECRET=whsec_... node webhook-receiver.mjs` or `FILEVERITY_WEBHOOK_SECRET=whsec_... python webhook_receiver.py`.
4. Choose **Send test** on the endpoint. Test keys never send events; **Send test** delivers a signed sample event instead.

**Receive webhooks: JavaScript**

```javascript
import { createHmac, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";

const secret = process.env.FILEVERITY_WEBHOOK_SECRET;
if (!secret) throw new Error("Set FILEVERITY_WEBHOOK_SECRET to the endpoint's signing secret.");
const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
const port = Number(process.env.PORT) || 3000;
const seen = new Set();

function verify(headers, body) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];
  if (!id || !timestamp || !signatures) return false;
  if (!(Math.abs(Date.now() / 1000 - Number(timestamp)) <= 300)) return false;
  const expected = createHmac("sha256", key).update(`${id}.${timestamp}.`).update(body).digest();
  return signatures.split(" ").some((entry) => {
    const [version, value = ""] = entry.split(",");
    const given = Buffer.from(value, "base64");
    return version === "v1" && given.length === expected.length && timingSafeEqual(given, expected);
  });
}

function handleEvent(event) {
  const { type, data } = event;
  if (data.test) {
    console.log(`Test event ${event.id}`);
  } else if (type === "analysis.completed") {
    const { level, action } = data.summary;
    console.log(`${data.id} succeeded: level ${level}, action ${action}, ${data.report_url}`);
  } else if (type === "analysis.failed") {
    console.log(`${data.id} ${data.status}: ${data.error.code}`);
  } else if (type === "balance.low") {
    const money = (cents) => `US$${(cents / 100).toFixed(2)}`;
    console.log(`Balance ${money(data.balance)} is below your ${money(data.threshold)} alert`);
  }
}

async function receive(req, res) {
  if (req.method !== "POST" || req.url !== "/webhooks/fileverity") return res.writeHead(404).end();
  const chunks = [];
  let size = 0;
  for await (const chunk of req) {
    size += chunk.length;
    if (size > 1_000_000) return res.writeHead(413).end();
    chunks.push(chunk);
  }
  const body = Buffer.concat(chunks);
  if (!verify(req.headers, body)) return res.writeHead(400).end();
  const event = JSON.parse(body.toString("utf8"));
  try {
    if (!seen.has(event.id)) {
      handleEvent(event);
      seen.add(event.id);
    }
  } catch (error) {
    console.error(`Could not handle ${event.id}`, error);
    return res.writeHead(500).end();
  }
  res.writeHead(200).end();
}

createServer((req, res) => receive(req, res).catch(() => res.destroy())).listen(port, () => console.log(`Listening on port ${port}`));
```

**Receive webhooks: Python**

```python
import base64
import hashlib
import hmac
import json
import os
import threading
import time

from flask import Flask, request

SECRET = os.environ.get("FILEVERITY_WEBHOOK_SECRET")
if not SECRET:
    raise SystemExit("Set FILEVERITY_WEBHOOK_SECRET to the endpoint's signing secret.")
KEY = base64.b64decode(SECRET.removeprefix("whsec_"))
seen = set()
lock = threading.Lock()
app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 1_000_000


def verify(headers, body):
    message_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp", "")
    signatures = headers.get("webhook-signature", "")
    if not message_id or not timestamp.isdecimal() or abs(time.time() - int(timestamp)) > 300:
        return False
    signed = f"{message_id}.{timestamp}.".encode() + body
    expected = hmac.new(KEY, signed, hashlib.sha256).digest()
    for entry in signatures.split(" "):
        version, _, value = entry.partition(",")
        try:
            given = base64.b64decode(value, validate=True)
        except ValueError:
            continue
        if version == "v1" and hmac.compare_digest(given, expected):
            return True
    return False


def handle_event(event):
    data = event["data"]
    if data.get("test"):
        print(f"Test event {event['id']}")
    elif event["type"] == "analysis.completed":
        level, action = data["summary"]["level"], data["summary"]["action"]
        print(f"{data['id']} succeeded: level {level}, action {action}, {data['report_url']}")
    elif event["type"] == "analysis.failed":
        print(f"{data['id']} {data['status']}: {data['error']['code']}")
    elif event["type"] == "balance.low":
        balance, threshold = data["balance"] / 100, data["threshold"] / 100
        print(f"Balance US${balance:.2f} is below your US${threshold:.2f} alert")


@app.post("/webhooks/fileverity")
def receive():
    body = request.get_data()
    if not verify(request.headers, body):
        return "", 400
    event = json.loads(body)
    with lock:
        try:
            if event["id"] not in seen:
                handle_event(event)
                seen.add(event["id"])
        except Exception:
            app.logger.exception("Could not handle %s", event["id"])
            return "", 500
    return "", 200


if __name__ == "__main__":
    port = int(os.environ.get("PORT", "3000"))
    print(f"Listening on port {port}")
    app.run(port=port)
```

**Output of the receiver**

```text
Listening on port 3000
Test event evt_Qm4vB7nP1sK8tL3wR6yZ2cXa
an_8Hk2pQ4vN9sL1xR7cW3mY6zB succeeded: level L2, action accept, https://fileverity.com/verify/FV-7K2M-9QXD-4T8R
```

- **Raw body.** The signature covers the exact bytes that were sent. The programs read the body as bytes, compute an HMAC-SHA256 over `webhook-id`, a dot, `webhook-timestamp`, a dot and those bytes with the base64-decoded secret as the key, and parse the JSON only after the signature is valid.
- **Constant-time comparison.** `timingSafeEqual` in Node.js and `hmac.compare_digest` in Python compare the result with every `v1` signature in the `webhook-signature` header.
- **Timestamp tolerance.** A `webhook-timestamp` more than five minutes from the clock, or one that is not a number, is rejected.
- **Fast answers.** A valid delivery gets `200` as soon as the event is handled. A bad signature gets `400`. If handling fails, the program answers `500`, and Fileverity retries later with the same event ID. See [Delivery and retries](https://fileverity.com/en/docs/webhooks.md#delivery).
- **Duplicates.** Delivery is at least once, and a retry keeps the event ID. The programs remember the IDs they have handled and answer `200` without handling an event twice. In production, replace the in-memory set with a unique constraint on the event ID in your database, and put the work on a queue so the answer stays fast.
- **Event data.** `analysis.completed` and `analysis.failed` carry the analysis without `result`; call `GET /analyses/{id}` when you need the full report. `balance` and `threshold` in `balance.low` are in US cents.

## Where to go next

- Run a whole intake flow with human review in [Document checks in an intake workflow](https://fileverity.com/en/docs/intake-workflow.md).
- Look up any field in the [API reference](https://fileverity.com/en/docs/api-reference.md), or load the OpenAPI document into your own tools.
- Go live with a Live key, a monthly spend limit and a `balance.low` webhook. See [Pricing and billing](https://fileverity.com/en/docs/pricing.md#alerts).
- Need help? Quote the `x-request-id` header when you [contact support](https://fileverity.com/en/contact).