Skip to content

python.mdx

Latest commit

 

History

History
999 lines (776 loc) · 40.7 KB

python.mdx

File metadata and controls

999 lines (776 loc) · 40.7 KB
Raw
title Arcjet Python SDK reference
prev false
generateMarkdownRoute true

import { Code, TabItem, Tabs } from "@astrojs/starlight/components"; import WhatIsArcjet from "/src/components/WhatIsArcjet.astro"; import Comments from "/src/components/Comments.astro"; import { Link } from "@/components/link"; import ConfigurationAsync from "/src/snippets/reference/python/Configuration.py?raw"; import ConfigurationSync from "/src/snippets/reference/python/ConfigurationSync.py?raw"; import MultipleRules from "/src/snippets/reference/python/MultipleRules.py?raw"; import Proxies from "/src/snippets/reference/python/Proxies.py?raw"; import ProtectAsync from "/src/snippets/reference/python/Protect.py?raw"; import ProtectSync from "/src/snippets/reference/python/ProtectSync.py?raw"; import WithRuleAsync from "/src/snippets/reference/python/WithRule.py?raw"; import WithRuleSync from "/src/snippets/reference/python/WithRuleSync.py?raw"; import IPLocation from "/src/snippets/reference/python/IPLocation.py?raw"; import ErrorLogging from "/src/snippets/reference/python/ErrorLogging.py?raw"; import Guard from "/src/snippets/reference/python/Guard.py?raw"; import Requirements from "@/snippets/shared/python/Requirements.mdx";

PyPI

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

Installation

Install from PyPI with your preferred package manager:

{/* prettier-ignore */} 

uv add arcjet
```sh 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

Quick start

See the <Link.ToSdk href="/get-started" sdk="python">quick start guide</Link.ToSdk>.

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 <Link.Page href="/shield">shield</Link.Page>, <Link.Page href="/rate-limiting">rate limiting</Link.Page>, <Link.Page href="/bot-protection">bot protection</Link.Page>, <Link.Page href="/email-validation">email validation</Link.Page>, <Link.Page href="/prompt-injection">prompt injection detection</Link.Page>, <Link.Page href="/sensitive-info">sensitive information detection</Link.Page>, <Link.Page href="/filters">request filters</Link.Page>.

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 <Link.Page href="/environment#arcjet-env">ARCJET_ENV</Link.Page> environment variable. Pass this when your config library doesn't propagate .env into os.environ (for example, pydantic-settings). See <Link.Page href="/reference/python#pydantic-settings-users">Pydantic-settings users</Link.Page>.
  • 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.

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 <Link.Page href="/bot-protection/reference">Bot protection</Link.Page> and <Link.Page href="/email-validation/reference">Email validation</Link.Page>.

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.

Environment variables

The Arcjet Python SDK uses several environment variables to configure its behavior. For more information, see <Link.Page href="/environment">Concepts: Environment variables</Link.Page>. 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 <Link.Page href="/environment#arcjet-env">ARCJET_ENV</Link.Page> 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.

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.

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.

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 <Link.Page href="/inspect">decision inspection reference</Link.Page> 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 <Link.Page href="/rate-limiting/reference#rate-limit-headers">rate limiting reference</Link.Page>.

See the <Link.Page href="/shield">shield</Link.Page>, <Link.Page href="/bot-protection">bot protection</Link.Page>, <Link.Page href="/rate-limiting">rate limiting</Link.Page>, and <Link.Page href="/email-validation">email validation</Link.Page> 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

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 <Link.Page href="/environment#arcjet-env">ARCJET_ENV</Link.Page>) 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.

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 <Link.Page href="/guards">Guards documentation</Link.Page>.

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.

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.
  • Official CrewAI tool calls – register_arcjet_hooks (arcjet.guard.crewai, crewai>=1.15.3,<2). There is no arcjet[crewai] extra. The gate is process-wide PRE_TOOL_CALL plus HookAborted(reason=..., source="arcjet"). guard_tool wraps a standalone BaseTool you call yourself. POST_TOOL_CALL is not registered.

LangChain checkpoint helpers and CrewAI guard_tool record metadata.outcome on the capture event. The value is capture and Sequence metadata. It doesn't change conclusion, which remains ALLOW or DENY. For the five values and the degraded discriminator, see <Link.Page href="/guards/capture#capture-outcomes">Capture outcomes</Link.Page>. register_arcjet_hooks records success when the action proceeds. It doesn't record degraded.

For install, examples, correlation, and the configure-before-wrap rule, see the <Link.Page href="/guards/langchain">LangChain agent guard</Link.Page>. For CrewAI hooks, see the <Link.Page href="/guards/crewai">CrewAI agent guard</Link.Page>.

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 <Link.Page href="/content-moderation">Content moderation</Link.Page>.

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 <Link.Page href="/guards/reference#metadata">Guard metadata</Link.Page>.

Record what happened with capture()

guard() decides whether something is allowed. <Link.Page href="/guards/capture">capture()</Link.Page> records that it happened. It never affects a decision, never raises, and is not awaited even on the async client. LangChain checkpoint helpers and CrewAI guard_tool set metadata.outcome on the events they emit. A direct capture() call doesn't. See <Link.Page href="/guards/capture#capture-outcomes">Capture outcomes</Link.Page>.

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 <Link.Page href="/testing#test-guard-and-capture-calls">registering a client and the test client</Link.Page> 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 <Link.Page href="/testing#test-guard-and-capture-calls">Testing Arcjet</Link.Page>.

Version support

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

<Link.Page href="/support">Technical support</Link.Page> 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.