Arcjet Python SDK reference

This is the reference guide for the Arcjet Python SDK, available on GitHub and licensed under the Apache 2.0 license.

What is Arcjet?

Arcjet is the runtime security platform that ships with your code. Enforce budgets, stop prompt injection, detect bots, and protect personal information with Arcjet's AI security building blocks.

Installation

Install from PyPI with your preferred package manager:

uv add arcjet

pip install arcjet

Prefer a glibc Linux container image such as python:3.10-slim or astral/uv:python3.10-trixie-slim. Alpine/musl isn’t a supported install target.

Requirements

CPython 3.10 or later on macOS, Windows, and glibc Linux (Debian, Ubuntu, RHEL, and *-slim / manylinux container images)
Alpine Linux and other musl-based systems aren’t supported. Prefer a glibc container image such as python:3.10-slim or astral/uv:python3.10-trixie-slim

Two runtime dependencies ship native code:

wasmtime: local WASM rule evaluation
pyqwest: HTTP client used by connect-python

Those packages publish musllinux wheels, so pip install arcjet can succeed on Alpine without a compiler. Alpine/musl isn’t a supported target. A missing wheel forces a source build that needs Rust and a C toolchain.

Quick start

See the quick start guide.

Protect versus Guard

The Arcjet Python SDK has two entrypoints. Pick the one that matches the surface you need to protect:

Arcjet Protect – arcjet (async) and arcjet_sync (sync). Protect HTTP request handlers in FastAPI, Flask, Django, and other Python web frameworks. You pass the framework request object to protect() and get an ArcjetDecision back. This is what you want for route handlers and API endpoints.
Arcjet Guard – arcjet.guard (with launch_arcjet / launch_arcjet_sync). Apply security rules where HTTP middleware can’t reach: AI agent tool calls, MCP servers, queue consumers, and background jobs. There is no request object – you pass inputs directly to guard().

	Protect (`arcjet` / `arcjet_sync`)	Guard (`arcjet.guard`)
Designed for	HTTP request protection	AI agent tool calls, background jobs
Request object	Required (`protect(request, ...)`)	Not needed
Rule binding	Rules configured once, input through `protect()` kwargs	Rules configured as classes, called with input per invocation
Rate limit key	IP or `characteristics` dict	Explicit `key` string (SHA-256 hashed before sending)
Rate limiting	✅	✅
Prompt injection detection	✅	✅
Content moderation	–	✅
Sensitive information detection	✅	✅
Bot protection	✅	–
Shield WAF	✅	–
Email validation	✅	–
Request filters	✅	–
IP analysis	✅	–
Custom rules	–	✅

Both entrypoints ship in the arcjet package – no extra install is required.

Protect

Use arcjet (async) or arcjet_sync (sync) to protect HTTP route handlers.

Async versus sync client

The SDK ships two clients with an identical API:

arcjet – async client for use with FastAPI and other async frameworks. Call await aj.protect(...).
arcjet_sync – sync client for use with Flask, Django, and other sync frameworks. Call aj.protect(...).

Pick the one that matches your framework. The rest of this section shows both where the API differs.

Configuration

Create a new Arcjet client with your API key and rules. Create it at startup, outside of the request handler.

The following fields are required:

key (str) – Your Arcjet site key. This can be found in the SDK Installation section for the site in the Arcjet Dashboard.
rules – The rules to apply to the request. See the various sections of the docs for how to configure these, such as shield, rate limiting, bot protection, email validation, prompt injection detection, sensitive information detection, request filters.

The following fields are optional:

proxies (list[str]) – A list of one or more trusted proxies. Arcjet excludes these addresses when it determines the client IP address. This is useful if you are behind a load balancer or proxy that sets the client IP address in a header. For an example, see Load balancers and proxies.
environment (str | None) – Explicit development/production mode ("development" or "production"). When None (default), falls back to the ARCJET_ENV environment variable. Pass this when your config library doesn’t propagate .env into os.environ (for example, pydantic-settings). See Pydantic-settings users.
disable_automatic_ip_detection (bool) – Disable automatic client IP detection so the application can provide ip_src to every protect() call. Defaults to False. This option cannot be combined with proxies.
timeout_ms (int) – Request timeout in milliseconds. Defaults to 2000 ms for every rule, in both development and production. An explicit timeout_ms overrides the default.

FastAPI (async)
Flask (sync)

import os

from arcjet import Mode, arcjet, shield

aj = arcjet(
    # Get your site key from https://app.arcjet.com and set it as an
    # environment variable rather than hard coding it.
    key=os.environ["ARCJET_KEY"],
    rules=[
        # Protect against common attacks with Arcjet Shield
        shield(mode=Mode.LIVE),  # Use Mode.DRY_RUN to log only
    ],
)

import os

from arcjet import Mode, arcjet_sync, shield

aj = arcjet_sync(
    # Get your site key from https://app.arcjet.com and set it as an
    # environment variable rather than hard coding it.
    key=os.environ["ARCJET_KEY"],
    rules=[
        # Protect against common attacks with Arcjet Shield
        shield(mode=Mode.LIVE),  # Use Mode.DRY_RUN to log only
    ],
)

Single instance

We recommend creating a single instance of the Arcjet client and reusing it throughout your application. This is because the SDK caches decisions and configuration to improve performance.

# Good — one instance, created once at startup
aj = arcjet(key=arcjet_key, rules=[...])

# Bad — new instance per request wastes resources
@app.get("/")
async def index(request: Request):
    aj = arcjet(key=arcjet_key, rules=[...])  # don't do this

Rule modes

Each rule can be configured in either Mode.LIVE or Mode.DRY_RUN. When in DRY_RUN mode, each rule returns its decision, but the end conclusion is always ALLOW.

This lets you run Arcjet in passive or demo mode to test rules before enabling them.

HTTP Protect rule factories require mode. Omitting it raises TypeError. This differs from JavaScript HTTP rules, which default to "DRY_RUN". Guard constructors still default to Mode.LIVE.

Pass mode on shield(), detect_bot(), token_bucket(), fixed_window(), sliding_window(), validate_email(), detect_sensitive_info(), filter_request(), and detect_prompt_injection(). protect_signup() forwards nested rate_limit, bots, and email mappings to those factories, so each mapping must include mode too.

This requirement is on Python SDK main, not in published arcjet 0.9.0 or 0.10.0b1.

from arcjet import Mode, detect_bot

detect_bot(mode=Mode.DRY_RUN, allow=[])

detect_bot and validate_email require exactly one of allow or deny. The BotDetection and EmailValidation dataclasses enforce the same requirement. An empty allow list is valid: it blocks every detected bot, or allows no email types. Passing neither list or both lists raises ValueError. For more information about these rules, see Bot protection and Email validation.

Because the top level conclusion is always ALLOW in DRY_RUN mode, you can loop through each rule result to check what would have happened:

for result in decision.results:
    if result.is_denied():
        print("Rule returned deny conclusion", result)

Multiple rules

You can combine rules to create a more complex protection strategy. For example, you can combine rate limiting and bot protection rules to protect your API from automated clients.

Declaration order does not control which Mode.LIVE deny you see. Local WebAssembly evaluation sorts rules by the same priority table as the JS and Go Protect SDKs. The first Mode.LIVE deny stops evaluation of later local rules.

The following table lists the local evaluation order, from first to last:

Priority	Rule	Constructor
1	Sensitive information	`detect_sensitive_info`
2	Filter	`filter_request`
3	Shield	`shield`
4	Rate limiting	`token_bucket`, `fixed_window`, `sliding_window`
5	Bot protection	`detect_bot`
6	Email validation	`validate_email`
7	Prompt injection	`detect_prompt_injection`

Rules with the same priority keep their declaration order. The three rate-limit constructors share priority 4. Unknown rule types sort last, at priority 100.

Sensitive information detection runs first so a Mode.LIVE deny happens before another rule can forward the payload.

detect_prompt_injection is listed so the table matches the JS SDK. The Python SDK does not evaluate prompt injection locally. The rank is reserved.

import os

from arcjet import Mode, arcjet, detect_bot, token_bucket

aj = arcjet(
    key=os.environ["ARCJET_KEY"],
    rules=[
        # Create a token bucket rate limit. Other algorithms are supported
        token_bucket(
            mode=Mode.LIVE,  # Use Mode.DRY_RUN to log only
            refill_rate=5,  # Refill 5 tokens per interval
            interval=10,  # Refill every 10 seconds
            capacity=10,  # Bucket capacity of 10 tokens
        ),
        # Detect automated clients
        detect_bot(
            mode=Mode.LIVE,
            allow=[],  # An empty allow list blocks all bots
        ),
    ],
)

Environment variables

The Arcjet Python SDK uses several environment variables to configure its behavior. For more information, see Concepts: Environment variables. The ARCJET_KEY environment variable is not read automatically: pass it explicitly with the key argument.

Pydantic-settings users

If you use pydantic-settings, pass ARCJET_ENV through the environment= kwarg. By design, pydantic-settings loads .env into a typed BaseSettings object rather than writing values back to os.environ. The SDK reads ARCJET_ENV with os.getenv, so it doesn’t pick up the value through that channel. Without the kwarg, the SDK defaults to production mode.

from arcjet import arcjet
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")

    ARCJET_KEY: str
    ARCJET_ENV: str = "development"


settings = Settings()

aj = arcjet(
    key=settings.ARCJET_KEY,
    rules=[...],
    environment=settings.ARCJET_ENV,
)

arcjet_sync() accepts the same kwarg.

Load balancers and proxies

If your application is behind a load balancer, Arcjet sees only the IP address of the load balancer and not the real client IP address.

To fix this, most load balancers set the X-Forwarded-For header with the real client IP address plus a list of proxies that the request has passed through.

The problem is that the client can spoof the X-Forwarded-For header, so trust it only if you are sure the load balancer sets it correctly. For more information, see the MDN documentation for X-Forwarded-For.

You can configure Arcjet to trust IP addresses in the X-Forwarded-For header by setting the proxies field in the configuration. Set this to a list of the IP addresses or CIDR ranges of your load balancers to remove, so the last IP address in the list is the real client IP address.

Example

For example, if the load balancer is at 203.0.113.100 and the client IP address is 198.51.100.1, the X-Forwarded-For header is:

X-Forwarded-For: 198.51.100.1, 203.0.113.100

Set the proxies field to ["203.0.113.100"] so Arcjet uses 198.51.100.1 as the client IP address.

You can also specify CIDR ranges to match multiple IP addresses.

import os

from arcjet import arcjet

aj = arcjet(
    key=os.environ["ARCJET_KEY"],
    rules=[],
    proxies=[
        "203.0.113.100",  # A single IP
        "203.0.113.0/24",  # A CIDR for the range
    ],
)

Ad hoc rules

Sometimes it is useful to add extra protection with a rule based on the logic in your handler. You usually want to inherit the rules, cache, and other configuration from the primary client. Use with_rule() on Arcjet or ArcjetSync for that.

with_rule() accepts a single rule or a sequence of rules. It returns a new client. The clone shares this instance’s DecisionCache, key, characteristics, and transport. The original client is unchanged.

You can call with_rule() more than once to add rules incrementally.

This method is on Python SDK main. It is not in published arcjet 0.9.0 or 0.10.0b1.

FastAPI (async)
Flask (sync)

import os

from arcjet import Mode, arcjet, detect_bot, fixed_window, shield
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

aj = arcjet(
    key=os.environ["ARCJET_KEY"],
    rules=[
        # Protect against common attacks with Arcjet Shield
        shield(mode=Mode.LIVE),  # Use Mode.DRY_RUN to log only
    ],
)


def get_client(user_id: str | None):
    if user_id:
        return aj
    # Only apply bot detection and rate limiting to guests
    return aj.with_rule(
        [
            fixed_window(mode=Mode.LIVE, window=60, max=10),
            detect_bot(mode=Mode.LIVE, allow=[]),  # empty allow blocks all bots
        ]
    )


@app.get("/")
async def index(request: Request):
    # Replace with a session lookup that returns the authenticated user ID
    user_id = "totoro"

    decision = await get_client(user_id).protect(request)

    if decision.is_denied():
        if decision.reason_v2.type == "RATE_LIMIT":
            return JSONResponse({"error": "Too Many Requests"}, status_code=429)
        return JSONResponse({"error": "Forbidden"}, status_code=403)

    return {"message": "Hello world"}

import os

from arcjet import Mode, arcjet_sync, detect_bot, fixed_window, shield
from flask import Flask, jsonify, request

app = Flask(__name__)

aj = arcjet_sync(
    key=os.environ["ARCJET_KEY"],
    rules=[
        # Protect against common attacks with Arcjet Shield
        shield(mode=Mode.LIVE),  # Use Mode.DRY_RUN to log only
    ],
)


def get_client(user_id: str | None):
    if user_id:
        return aj
    # Only apply bot detection and rate limiting to guests
    return aj.with_rule(
        [
            fixed_window(mode=Mode.LIVE, window=60, max=10),
            detect_bot(mode=Mode.LIVE, allow=[]),  # empty allow blocks all bots
        ]
    )


@app.get("/")
def index():
    # Replace with a session lookup that returns the authenticated user ID
    user_id = "totoro"

    decision = get_client(user_id).protect(request)

    if decision.is_denied():
        if decision.reason_v2.type == "RATE_LIMIT":
            return jsonify(error="Too Many Requests"), 429
        return jsonify(error="Forbidden"), 403

    return jsonify(message="Hello world")

`protect()`

Arcjet exposes a single protect method that is used to execute your protection rules. It accepts the framework request object as its first argument. Rules you add to the SDK may require additional keyword arguments, such as the validate_email rule requiring an email argument.

The async client returns a coroutine that resolves to an ArcjetDecision object. The sync client returns the ArcjetDecision directly.

FastAPI (async)
Flask (sync)

import os

from arcjet import Mode, arcjet, token_bucket
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

aj = arcjet(
    key=os.environ["ARCJET_KEY"],
    rules=[
        # Create a token bucket rate limit. Other algorithms are supported
        token_bucket(
            mode=Mode.LIVE,
            characteristics=["userId"],  # Track requests by a custom user ID
            refill_rate=5,  # Refill 5 tokens per interval
            interval=10,  # Refill every 10 seconds
            capacity=10,  # Bucket capacity of 10 tokens
        ),
    ],
)


@app.get("/")
async def index(request: Request):
    user_id = "user_123"  # Replace with your authenticated user ID

    # The "userId" characteristic value is required because it is defined in
    # the characteristics field of the token_bucket rule.
    decision = await aj.protect(
        request,
        requested=5,  # Deduct 5 tokens from the bucket
        characteristics={"userId": user_id},
    )

    if decision.is_denied():
        return JSONResponse({"error": "Too Many Requests"}, status_code=429)

    return {"message": "Hello world"}

import os

from arcjet import Mode, arcjet_sync, token_bucket
from flask import Flask, jsonify, request

app = Flask(__name__)

aj = arcjet_sync(
    key=os.environ["ARCJET_KEY"],
    rules=[
        token_bucket(
            mode=Mode.LIVE,
            characteristics=["userId"],
            refill_rate=5,
            interval=10,
            capacity=10,
        ),
    ],
)


@app.get("/")
def index():
    user_id = "user_123"  # Replace with your authenticated user ID

    decision = aj.protect(
        request,
        requested=5,
        characteristics={"userId": user_id},
    )

    if decision.is_denied():
        return jsonify(error="Too Many Requests"), 429

    return jsonify(message="Hello world")

Parameters

These keyword arguments are optional unless required by a configured rule or client mode:

Parameter	Type	Used by
`requested`	`int`	Token bucket rate limit
`characteristics`	`Mapping[str, Any]`	Rate limiting (pass values for keys declared in rule config)
`detect_prompt_injection_message`	`str`	Prompt injection detection
`sensitive_info_value`	`str`	Sensitive info detection
`email`	`str`	Email validation
`filter_local`	`Mapping[str, str]`	Request filters (`local.*` fields)
`extra`	`Mapping[str, str]`	SDK-derived request context forwarded as a flat string map. Prefer `metadata` for application data.
`metadata`	`Metadata \| None`	Nested JSON for correlation and analytics. See Metadata.
`ip_src`	`str`	Manual IP override (advanced)
`correlation_id`	`str`	Correlates this decision with a guard call, workflow run, or agent trace. A dedicated, indexable field – not `extra` or `metadata` – and does not affect the decision or its cache key (`arcjet` >= 0.9.0)

HTTP detect_prompt_injection accepts only mode, which is required. On Python SDK main, omitting mode or passing threshold= raises TypeError. Guard DetectPromptInjection still defaults to LIVE.

Override the client IP

Arcjet normally detects the client IP address from the framework request. If your application has already determined the client IP from a trusted source, disable automatic detection when creating the client and pass ip_src to every protect() call:

aj = arcjet(
    key=arcjet_key,
    rules=[...],
    disable_automatic_ip_detection=True,
)

ip_src = get_client_ip_from_trusted_source(request)
decision = await aj.protect(request, ip_src=ip_src)

The sync client uses the same options without await. When automatic detection is disabled, omitting ip_src or passing an empty string raises an ArcjetMisconfiguration. Passing a non-empty ip_src while automatic detection is enabled also raises an ArcjetMisconfiguration. With the default automatic detection enabled and ip_src omitted, Arcjet detects the IP from the framework request.

Caution: The SDK trusts ip_src without validating it. Validate the value and ensure it comes from a trusted source. Do not pass a client-controlled header directly; doing so could allow clients to choose the IP address used for fingerprinting, rate limiting, and other security checks.

Metadata

protect() accepts metadata: a mapping of string keys to any JSON-serializable value, including nested objects and arrays. Prefer it over extra, which stays a flat Mapping[str, str] of SDK-derived request context.

decision = await aj.protect(
    request,
    metadata={
        "request_id": request_id,
        "user": {"id": user_id, "plan": "pro"},
        "flags": {"beta": True},
    },
)

See Metadata for limits, AJ1017 drop warnings, and the difference between Guard and protect() warning channels.

Decision

The protect method returns an ArcjetDecision object. It includes the following properties:

conclusion ("ALLOW" | "DENY" | "CHALLENGE" | "ERROR") – The final conclusion based on evaluating each of the configured rules.
reason_v2 – A typed reason object describing the conclusion. Use reason_v2.type as a discriminator ("BOT", "RATE_LIMIT", "SHIELD", "EMAIL", "SENSITIVE_INFO", "PROMPT_INJECTION", "FILTER", or "ERROR") and then access type-specific fields.
results – A list of per-rule result objects. There is one for each configured rule, so you can inspect the individual results.
ip / ip_details – Objects containing Arcjet’s analysis of the client IP address. For more information, see IP analysis.

Conclusion

Use the following ArcjetDecision methods to check the conclusion:

is_allowed() (bool) – Arcjet concluded that the request is allowed.
is_denied() (bool) – Arcjet concluded that the request is denied.
is_error() (bool) – There was an unrecoverable error.

The conclusion is the highest-severity finding from the configured rules. "DENY" is the highest severity, followed by "CHALLENGE", then "ERROR", and finally "ALLOW" as the lowest severity.

For example, when a bot protection rule returns an error and a validate email rule returns a deny, the overall conclusion would be deny. To access the error you would have to iterate over the results property on the decision.

Reason

The reason_v2 property of the ArcjetDecision object describes the conclusion. It always reflects the highest-priority rule that produced that conclusion; to inspect other rules, iterate over the results property on the decision. Local evaluation ranks rules as described in Multiple rules.

Switch on reason_v2.type to map each rule kind to a response. Only branch on reasons that produce a different response – a branch that returns 403 for SHIELD when the default already returns 403 is dead code.

if decision.is_denied():
    if decision.reason_v2.type == "RATE_LIMIT":
        return JSONResponse({"error": "Too many requests"}, status_code=429)
    if decision.reason_v2.type in ("EMAIL", "SENSITIVE_INFO", "PROMPT_INJECTION"):
        return JSONResponse({"error": "Bad request"}, status_code=400)
    # BOT, SHIELD, FILTER, and anything else
    return JSONResponse({"error": "Forbidden"}, status_code=403)

Recommended HTTP status mapping:

`reason_v2.type`	Status
`"RATE_LIMIT"`	429
`"EMAIL"`	400
`"SENSITIVE_INFO"`	400
`"PROMPT_INJECTION"`	400
`"BOT"`, `"SHIELD"`, `"FILTER"`, fallback	403

Each variant exposes type-specific fields:

`reason_v2.type`	Fields
`"BOT"`	`allowed`, `denied`, `spoofed` (`bool`), `verified` (`bool`)
`"RATE_LIMIT"`	`max`, `remaining`, `reset_time`, `reset`, `window`
`"SHIELD"`	`shield_triggered` (`bool`)
`"EMAIL"`	`email_types` (for example, `["DISPOSABLE", "NO_MX_RECORDS"]`)
`"SENSITIVE_INFO"`	`allowed`, `denied` (each a list of `IdentifiedEntity`)
`"PROMPT_INJECTION"`	`injection_detected` (`bool`); `score` (`float`, deprecated)
`"FILTER"`	`matched_expressions`, `undetermined_expressions`
`"ERROR"`	`message` (`str`)

Results

The results property contains a list of per-rule result objects. There is one for each configured rule, so you can inspect the individual results.

for result in decision.results:
    print("Rule Result", result)

Each result includes:

conclusion – The conclusion of the rule ("ALLOW", "DENY", "CHALLENGE", or "ERROR").
reason_v2 – A typed reason for this rule’s conclusion (same set of types as on the top-level decision).
is_denied() / is_allowed() / is_error() – convenience methods.

For bot results, the SDK exports helpers that match the JavaScript @arcjet/inspect utilities. Import them from arcjet. See the decision inspection reference for return values.

from arcjet import (
    is_missing_user_agent,
    is_spoofed_bot,
    is_verified_bot,
    set_rate_limit_headers,
)

if any(is_verified_bot(r) for r in decision.results):
    return jsonify(message="Hello bot")

if any(is_spoofed_bot(r) for r in decision.results):
    return jsonify(error="Spoofed bot"), 403

if any(is_missing_user_agent(r) for r in decision.results):
    return jsonify(error="User-Agent required"), 400

set_rate_limit_headers(response, decision)

is_verified_bot, is_spoofed_bot, and is_missing_user_agent ignore "DRY_RUN" results.

set_rate_limit_headers writes IETF RateLimit and RateLimit-Policy headers onto a response, response.headers, or a mutable mapping. When several rate limit results are present, the tightest remaining budget is advertised. If two policies share the same max, no headers are written. For more information about rate limit headers, see the rate limiting reference.

See the shield, bot protection, rate limiting, and email validation docs for what each rule’s reason fields mean.

IP analysis

Arcjet returns IP metadata with every decision – no extra API calls needed.

# High-level helpers on decision.ip
if decision.ip.is_hosting():
    # likely a cloud / hosting provider — often suspicious for bots
    pass

if decision.ip.is_vpn() or decision.ip.is_proxy() or decision.ip.is_tor():
    # apply your policy for anonymized traffic
    pass

if decision.ip.is_abuser():
    # IP is associated with known abuse
    pass

# Typed field access via decision.ip_details
ip = decision.ip_details
if ip:
    print(ip.city, ip.country_name)   # geolocation
    print(ip.asn, ip.asn_name)        # ASN / network
    print(ip.is_vpn, ip.is_hosting)   # reputation

decision.ip exposes boolean helpers: is_hosting(), is_vpn(), is_proxy(), is_tor(), is_abuser().

decision.ip_details is an IpDetails dataclass (or None) with these fields:

Geolocation: latitude, longitude, accuracy_radius, timezone, postal_code, city, region, country, country_name, continent, continent_name.
Network (ASN): asn, asn_name, asn_domain, asn_type (one of isp, hosting, business, education), asn_country.
Reputation: is_vpn, is_proxy, is_tor, is_hosting, is_relay, is_abuser, service (for example, "Apple Private Relay").

The IP fields may be missing – decision.ip_details itself may be None, and individual fields may be None. Geolocation accuracy varies; country is usually reliable, but city and region can be very inaccurate. Use these fields for convenience (for example, suggesting a user location) but do not rely on them alone.

IP location example

import os

from arcjet import Mode, arcjet, shield
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

aj = arcjet(
    key=os.environ["ARCJET_KEY"],
    rules=[
        shield(mode=Mode.LIVE),
    ],
)


@app.get("/")
async def index(request: Request):
    decision = await aj.protect(request)

    if decision.is_denied():
        return JSONResponse({"error": "Forbidden"}, status_code=403)

    ip = decision.ip_details
    if ip and ip.country:
        return {
            "message": f"Hello {ip.country_name}!",
            "ip": {
                "country": ip.country,
                "country_name": ip.country_name,
                "continent": ip.continent,
                "continent_name": ip.continent_name,
                "asn": ip.asn,
                "asn_name": ip.asn_name,
                "asn_domain": ip.asn_domain,
            },
        }

    return {"message": "Hello world"}

For the IP address 8.8.8.8 you might get the following response. Arcjet returns only the fields it has data for:

{
  "message": "Hello United States!",
  "ip": {
    "country": "US",
    "country_name": "United States",
    "continent": "NA",
    "continent_name": "North America",
    "asn": "AS15169",
    "asn_name": "Google LLC",
    "asn_domain": "google.com"
  }
}

Arcjet automatically detects the IP address of the client making the request based on the context provided by your framework. In development (see ARCJET_ENV) we allow private and internal addresses so that the SDK works correctly locally.

Error handling

Arcjet is designed to fail open so that a service issue or misconfiguration does not block all requests. If there is an error condition when processing a rule, Arcjet returns an ERROR result for that rule and you can check result.reason_v2.message for more information.

If all other rules that were run returned an ALLOW result, then the final Arcjet conclusion is ERROR.

import logging
import os

from arcjet import Mode, arcjet, sliding_window
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

logger = logging.getLogger(__name__)

aj = arcjet(
    key=os.environ["ARCJET_KEY"],
    rules=[
        sliding_window(mode=Mode.LIVE, interval=3600, max=60),
    ],
)


@app.get("/")
async def index(request: Request):
    decision = await aj.protect(request)

    for result in decision.results:
        if result.reason_v2.type == "ERROR":
            # Fail open by logging the error and continuing
            logger.warning("Arcjet error: %s", result.reason_v2.message)
            # You could also fail closed here for very sensitive routes
            # return JSONResponse({"error": "Service unavailable"}, status_code=503)

    if decision.is_denied():
        return JSONResponse({"error": "Forbidden"}, status_code=403)

    return {"message": "Hello world"}

You can check for errors at the top level too:

decision = await aj.protect(request)

if decision.is_error():
    # Arcjet service error — fail open or apply a fallback policy
    pass
elif decision.is_denied():
    return JSONResponse({"error": "Denied"}, status_code=403)

Guard

arcjet.guard is a lower-level API designed for AI agent tool calls, MCP servers, and background tasks where there is no HTTP request object. It gives you fine-grained, per-call control over rate limiting, prompt injection detection, content moderation, sensitive information detection, and custom rules. For the full guide, see the Guards documentation.

Setup

Use launch_arcjet for async frameworks and launch_arcjet_sync for sync frameworks. Create a single client at startup and reuse it. Configure each rule once, then bind input and call guard() per invocation. The client request timeout defaults to 2000 ms (timeout_ms on launch_arcjet / launch_arcjet_sync), matching the JavaScript Guard default.

import os
import time

from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet

# Create a single guard client at startup and reuse it
aj = launch_arcjet(key=os.environ["ARCJET_KEY"])

# Configure rules once at module scope so per-rule result accessors work
user_limit = TokenBucket(
    refill_rate=100,
    interval_seconds=60,
    max_tokens=1000,
    bucket="user-tools",  # name this per use case to avoid collisions
)
prompt_scan = DetectPromptInjection()


async def handle_tool_call(user_id: str, message: str) -> str:
    # Bind input and call guard() for each invocation. Hardcode `label`
    # as a string literal so it stays greppable and groups in the dashboard.
    decision = await aj.guard(
        label="tools.weather",
        rules=[
            user_limit(key=user_id, requested=5),
            prompt_scan(message),
        ],
        metadata={"user_id": user_id},
    )

    if decision.conclusion == "DENY":
        # Branch on which rule denied to give the caller something actionable
        rate_limited = user_limit.denied_result(decision)
        if rate_limited:
            retry_in = max(0, rate_limited.reset_at_unix_seconds - int(time.time()))
            raise RuntimeError(f"Rate limited — retry in {retry_in}s")
        raise RuntimeError("Blocked")

    # Safe to proceed with the tool call
    return "..."

Checkpoint helpers

To wrap an effect instead of handling the decision yourself, use the checkpoint helpers. They fail closed by default (on_guard_error="deny"). A DENY raises ArcjetDeniedError or ArcjetToolDeniedError. Unavailability raises ArcjetUnavailableError or ArcjetToolUnavailableError.

Any Python callable – guard_action / guard_action_sync in arcjet.guard. No extra.
A LangChain BaseTool you call yourself – guard_tool (arcjet[langchain]).
An agent from create_agent – ArcjetMiddleware + ToolPolicy (arcjet[langchain-agents]).
Observe only – ArcjetCaptureHandler / ArcjetAsyncCaptureHandler. These cannot deny a call.

For install, examples, correlation, and the configure-before-wrap rule, see the LangChain agent guard.

Rules

Configure each rule once at module scope so you have a stable reference for the typed per-rule result accessors (for example, user_limit.denied_result(decision)). All rules accept keyword-only arguments. Every rule accepts mode ("LIVE" or "DRY_RUN"), label (an observability label that appears in the dashboard), and metadata (nested JSON – see Metadata).

Rate limiting

from arcjet.guard import TokenBucket, FixedWindow, SlidingWindow

user_limit = TokenBucket(
    refill_rate=10,
    interval_seconds=60,
    max_tokens=100,
    bucket="user-tools",  # name this per use case to avoid collisions
)

team_limit = FixedWindow(
    max_requests=1000,
    window_seconds=3600,
    bucket="team-api",
)

api_limit = SlidingWindow(
    max_requests=500,
    interval_seconds=60,
    bucket="public-api",
)

Rate limit state is tracked server-side by the combination of bucket and other configuration. Set bucket explicitly to avoid collisions between different rules – two rate limit rules created with the default bucket name share counters.

At call time, all three accept key=... (the per-caller identifier – user ID, session ID, tenant) and requested=N (tokens or requests consumed; default 1).

Prompt injection detection

from arcjet.guard import DetectPromptInjection

prompt_scan = DetectPromptInjection()

decision = await aj.guard(
    label="tools.weather",
    rules=[prompt_scan(user_message)],
)

Sensitive information detection

Runs locally in WebAssembly – the raw text never leaves the SDK; only a SHA-256 hash is sent alongside the local result. Valid entity types: "EMAIL", "PHONE_NUMBER", "IP_ADDRESS", "CREDIT_CARD_NUMBER".

from arcjet.guard import LocalDetectSensitiveInfo

sensitive = LocalDetectSensitiveInfo(
    deny=["EMAIL", "CREDIT_CARD_NUMBER"],
)

allow and deny are mutually exclusive.

Content moderation

Guard-only. Instantiate once, then bind the untrusted text per call. The result reports detected and optional billing (text_units) – not per-category scores. See Content moderation.

from arcjet.guard import ModerateContent

moderate = ModerateContent()

decision = await aj.guard(
    label="llm.output",
    rules=[moderate(text)],
)

if decision.conclusion == "DENY" and decision.reason == "MODERATE_CONTENT":
    raise RuntimeError("Harmful content detected")

result = moderate.result(decision)
if result:
    print(result.detected)
    if result.billing:
        print(result.billing.unit, result.billing.count)

Custom rules

Subclass LocalCustomRule and override evaluate (sync) or evaluate_async (async) to implement custom logic with typed Config / Input / Data shapes.

`guard()`

The guard call takes a label identifying the invocation site, a list of bound rule inputs, and optional metadata:

Parameter	Type	Description
`label`	`str`	Label identifying this guard call (required). Validated server-side as a slug – lowercase letters, digits, dash (`-`), and dot (`.`) only
`rules`	`Sequence[RuleWithInput]`	Bound rule inputs (required)
`metadata`	`Metadata \| None`	Nested JSON for correlation and analytics – see Metadata
`correlation_id`	`str \| None`	Optional ID correlating this decision with a request, workflow run, or agent trace. A dedicated field, not metadata; does not affect the decision (`arcjet` >= 0.9.0)

Guard decision

Errors and warnings mean opposite things about how much to trust a decision. An error means the security signal may be degraded: a rule couldn’t be evaluated, so Arcjet failed open. A warning means the signal is intact and the decision is reporting a diagnostic. The guard decision exposes the following:

conclusion – "ALLOW" or "DENY". Always check before proceeding.
has_failed_open() – True when the conclusion is "ALLOW" only because a rule (or the decision itself) could not be processed – that is, the security signal was degraded and Arcjet failed open. This is the fail-closed gate: deny on it where a degraded signal is unacceptable (arcjet >= 0.9.0).
error_results() – the errored rule results (each with a code / message) for logging (arcjet >= 0.9.0).
warnings – diagnostics about your request that don’t degrade the signal (for example, a stripped invalid metadata key). Informational only; never changes the conclusion.
results – per-rule outcomes.

has_error() is deprecated as of arcjet 0.9.0 (it conflated warnings with rule errors and now emits a DeprecationWarning). Use has_failed_open() for the fail-closed gate and warnings for diagnostics.

For useful error messages, branch on which rule denied – not just on DENY. Each rule defined at module scope exposes typed result accessors:

rule.result(decision) – the result for this rule, or None.
rule.denied_result(decision) – the result, but only if the rule denied the request. Returns None otherwise.
rule.error_result(decision) – the RuleResultError if this specific rule errored, else None. The mirror of denied_result for the fail-open case (arcjet >= 0.9.0).

import time

if decision.conclusion == "DENY":
    rate_limited = user_limit.denied_result(decision)
    if rate_limited:
        retry_in = max(
            0, rate_limited.reset_at_unix_seconds - int(time.time())
        )
        raise TaskBlocked(f"rate limited — retry in {retry_in}s")
    raise TaskBlocked("blocked")

For token bucket rate limits the denied result also exposes remaining_tokens, max_tokens, refill_rate, and refill_interval_seconds. Fixed and sliding window results expose remaining_requests, max_requests, and reset_at_unix_seconds.

Hardcode the label argument to guard() as a string literal (for example, "tools.get-weather", not f"tools.{name}"). Labels are validated server-side as slugs – lowercase letters, digits, dash (-), and dot (.) only – so underscores and uppercase are rejected. Hardcoded labels stay greppable and the dashboard groups by them. Pass metadata whenever you have useful auditing context – nested objects and arrays are accepted, and it shows up in the dashboard.

Metadata

guard(), protect(), and every Guard rule accept metadata: a mapping of string keys to any JSON-serializable value, including nested objects and arrays.

decision = await aj.guard(
    label="tools.weather",
    rules=[user_limit(key=user_id)],
    metadata={
        "user": {"id": user_id, "plan": "pro"},
        "tool_name": "get_weather",
        "duration_ms": 160,
        "success": True,
    },
)

Each top-level value is JSON-encoded by the SDK and stored verbatim, so exact integers survive. Server-enforced limits: 128 top-level keys, 4 KiB per serialized value, 10 levels of nesting, and key names limited to letters, digits, -, ., and _. Over a limit, that key is dropped.

Nothing here can fail a call or change a decision. Dropped keys are reported: server-side drops arrive on decision.warnings, one per key. Keys the SDK cannot encode (datetime, a set, NaN, a circular reference) are collected into a single AJ1017 warning naming them. For protect(), which has no warnings channel, that warning is logged at WARNING instead.

Metadata is untrusted and is not redacted – do not put secrets or PII in it. The SDK also drops keys once one request’s metadata exceeds 768 KiB in total. For the full limit table and language-specific notes, see Guard metadata.

Record what happened with `capture()`

guard() decides whether something is allowed. capture() records that it happened. It never affects a decision, never raises, and is not awaited even on the async client.

aj.capture(
    action="refund.issued",
    correlation_id=workflow_id,
    decision_id=decision.id,
    metadata={"amount_cents": 4999, "invoice": {"id": "inv_123"}},
)

Call await aj.flush() (async) or aj.flush() (sync) at shutdown so the final batch is sent. See registering a client and the test client when capture() is too deep to receive a handle.

Optional: Register a client

from arcjet.guard import launch_arcjet, register_arcjet

register_arcjet(launch_arcjet(key=os.environ["ARCJET_KEY"]))

Free guard(), capture(), and flush() then reach the registered client. If nothing is registered, free guard() fail-opens – ALLOW with has_failed_open() true. capture() drops the event silently.

capture() is one function for both client flavors. guard() / flush() must match: await guard(...) with launch_arcjet(), or guard_sync(...) with launch_arcjet_sync(). The wrong pair fail-opens and reports AJ3007.

Use arcjet.guard.testing.register_test_client() in tests. See Testing Arcjet.

Version support

Arcjet supports CPython 3.10 and later on macOS, Windows, and glibc Linux. Alpine/musl isn’t supported.

Technical support is provided for the current major version of the Arcjet SDK for all users and for the current and previous major versions for paid users. We provide security fixes for the current and previous major versions.

Arcjet Python SDK reference

Installation

Requirements

Quick start

Protect versus Guard

Protect

Async versus sync client

Configuration

Single instance

Rule modes

Multiple rules

Environment variables

Pydantic-settings users

Load balancers and proxies

Example

Ad hoc rules

protect()

Parameters

Override the client IP

Metadata

Decision

Conclusion

Reason

Results

IP analysis

IP location example

Error handling

Guard

Setup

Checkpoint helpers

Rules

Rate limiting

Prompt injection detection

Sensitive information detection

Content moderation

Custom rules

guard()

Guard decision

Metadata

Record what happened with capture()

Optional: Register a client

Version support

Discussion

`protect()`

`guard()`

Record what happened with `capture()`