# arcjet
[Arcjet](https://arcjet.com) 是随您的 AI 代码一起发布运行时安全平台。阻止机器人和自动化攻击，防止您的 AI 预算被耗尽、数据泄露或工具被滥用。
这是 [Arcjet](https://arcjet.com) 的 Python SDK。
## 开始使用
1. **获取 API 密钥** — [在 `app.arcjet.com` 注册](https://app.arcjet.com)。
2. **安装 SDK：**
```
pip install arcjet
# 或使用 uv
uv add arcjet
```
3. **设置环境变量：**
```
# .env 或 .env.local
ARCJET_KEY=ajkey_yourkey
```
4. **保护路由** — 查看 [AI 保护示例](#quick-start) 或
以下各个 [功能示例](#features)。
### 获取帮助
[加入我们的 Discord 服务器](https://arcjet.com/discord) 或 [寻求支持](https://docs.arcjet.com/support)。
- [文档](https://docs.arcjet.com) — 完整参考和指南
- [示例](https://github.com/arcjet/arcjet-py/tree/main/examples) — FastAPI
和 Flask 示例应用，包括 LangChain 集成
- [蓝图](https://docs.arcjet.com/blueprints) — 常见安全模式的配方
## 快速开始
使用提示注入检测、令牌预算速率限制和机器人保护来保护 AI 聊天端点：
```
# main.py
import os
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from arcjet import (
arcjet, # async client — use arcjet_sync for Flask and other sync frameworks
detect_bot,
detect_prompt_injection,
detect_sensitive_info,
shield,
token_bucket,
Mode,
SensitiveInfoEntityType,
)
app = FastAPI()
arcjet_key = os.getenv("ARCJET_KEY")
if not arcjet_key:
raise RuntimeError(
"ARCJET_KEY is required. Get one at https://app.arcjet.com"
)
# 创建单个 Arcjet 实例并在请求间复用。
# 如果使用 Flask 或其他同步框架，请改用 arcjet_sync。
aj = arcjet(
key=arcjet_key,
rules=[
# Detect and block prompt injection attacks in user messages
detect_prompt_injection(mode=Mode.LIVE),
# Block sensitive data (e.g. credit cards, PII) from reaching your LLM
detect_sensitive_info(
mode=Mode.LIVE,
deny=[
SensitiveInfoEntityType.CREDIT_CARD_NUMBER,
SensitiveInfoEntityType.EMAIL,
SensitiveInfoEntityType.PHONE_NUMBER,
],
),
# Rate limit by token budget — refill 100 tokens every 60 seconds
token_bucket(
characteristics=["userId"],
mode=Mode.LIVE,
refill_rate=100,
interval=60,
capacity=1000,
),
# Block automated clients and scrapers from your AI endpoints
detect_bot(
mode=Mode.LIVE,
allow=[], # empty = block all bots
),
# Protect against common web attacks (SQLi, XSS, etc.)
shield(mode=Mode.LIVE),
],
)
class ChatRequest(BaseModel):
message: str
@app.post("/chat")
async def chat(request: Request, body: ChatRequest):
userId = "user_123" # replace with real user ID from session
decision = await aj.protect(
request,
requested=5, # tokens consumed per request
characteristics={"userId": userId},
detect_prompt_injection_message=body.message, # scan for prompt injection
sensitive_info_value=body.message, # scan for PII
)
if decision.is_denied():
status = 429 if decision.reason_v2.type == "RATE_LIMIT" else 403
return JSONResponse({"error": "Denied"}, status_code=status)
# Safe to pass body.message to your LLM
return {"reply": "..."}
```
## 功能特性
- 🔒 [提示注入检测](#prompt-injection-detection) — 在到达您的 LLM 之前检测并阻止提示注入攻击。
- 🤖 [机器人保护](#bot-protection) — 阻止爬虫、凭证填充器和 AI 爬虫滥用您的端点。
- 🛑 [速率限制](#rate-limiting) — 令牌桶、固定窗口和滑动窗口算法；为每个用户设置 AI 令牌预算。
- 🕵️ [敏感信息检测](#sensitive-information-detection) — 阻止 PII、信用卡号和自定义模式进入您的 AI 管道。
- 🛡️ [护盾 WAF](#shield-waf) — 保护免受 SQL 注入、XSS 和其他常见 Web 攻击。
- 📧 [电子邮件验证](#email-validation) — 在注册时阻止一次性、无效和无法投递的地址。
- 📝 [注册表单保护](https://docs.arcjet.com/signup-protection) —
结合机器人保护、电子邮件验证和速率限制来保护您的注册表单。
- 🎯 [请求过滤器](#request-filters) — 针对 IP、路径、标头和自定义本地字段的基于表达式的规则。
- 🌐 [IP 分析](#ip-analysis) — 每笔请求均包含地理位置、ASN、VPN、代理、Tor 和托管检测。
- 🧩 [Arcjet Guard](#arcjet-guard) — 针对 AI 代理工具调用和后台任务的更低层 API，其中不存在 HTTP 请求。
### 哪些功能是您需要的？
| 如果您的应用程序有... | 推荐的功能 |
| ----------------------------- | ----------------------------------------------------------------------------- |
| LLM / AI 聊天端点 | 提示注入 + 敏感信息 + 令牌桶速率限制 + 机器人保护 + 护盾 |
| AI 代理工具调用 | [Arcjet Guard](#arcjet-guard) — 速率限制 + 提示注入 + 敏感信息 + 自定义规则 |
| 公共 API | 速率限制 + 机器人保护 + 护盾 |
| 注册 / 登录表单 | 电子邮件验证 + 机器人保护 + 速率限制（或 [注册保护](https://docs.arcjet.com/signup-protection)） |
| 内部 / 管理路由 | 护盾 + 请求过滤器（国家、VPN/代理阻断） |
| 任何 Web 应用程序 | 护盾 + 机器人保护（适用于所有应用程序的良好基线） |
所有功能都可以在单个 Arcjet 实例中组合。规则会一起评估 — 如果 **任何** 规则拒绝请求，`decision.is_denied()` 返回 `True`。在强制执行之前，使用 `Mode.DRY_RUN` 单独测试每条规则。
## 安装
使用 [uv](https://docs.astral.sh/uv/) 从 [PyPI](https://pypi.org/project/arcjet/) 安装：
```
# 使用 uv 项目
uv add arcjet
# 使用现有的 pip 管理项目
uv pip install arcjet
```
或使用 pip：
```
pip install arcjet
```
## 提示注入检测
检测并阻止提示注入攻击 — 在到达您的模型之前阻止用户劫持 LLM 行为的行为。
### FastAPI
```
from arcjet import arcjet, detect_prompt_injection, Mode
aj = arcjet(
key=arcjet_key,
rules=[
detect_prompt_injection(mode=Mode.LIVE),
],
)
@app.post("/chat")
async def chat(request: Request, body: ChatRequest):
decision = await aj.protect(
request,
detect_prompt_injection_message=body.message,
)
if decision.is_denied():
return JSONResponse({"error": "Prompt injection detected"}, status_code=403)
# safe to pass body.message to your LLM
```
### Flask
```
from arcjet import arcjet_sync, detect_prompt_injection, Mode
aj = arcjet_sync(
key=arcjet_key,
rules=[
detect_prompt_injection(mode=Mode.LIVE),
],
)
@app.route("/chat", methods=["POST"])
def chat():
body = request.get_json()
decision = aj.protect(request, detect_prompt_injection_message=body["message"])
if decision.is_denied():
return jsonify(error="Prompt injection detected"), 403
# safe to pass body["message"] to your LLM
```
您可以通过 `threshold` 参数（0.0–1.0，默认为 0.5）调整检测灵敏度。较高的值需要更强的信号来触发拒绝，从而减少误报，但可能会遗漏细微的攻击：
```
detect_prompt_injection(mode=Mode.LIVE, threshold=0.8)
```
有关更多详细信息，请参阅 [提示注入文档](https://docs.arcjet.com/prompt-injection)。
## 机器人保护
管理来自自动化客户端的流量。阻止爬虫、凭证填充器和 AI 爬虫，同时允许合法的机器人（如搜索引擎和监控器）。
### FastAPI
```
from arcjet import arcjet, detect_bot, Mode, BotCategory
aj = arcjet(
key=arcjet_key,
rules=[
detect_bot(
mode=Mode.LIVE,
allow=[
BotCategory.SEARCH_ENGINE, # Google, Bing, etc.
# BotCategory.MONITOR, # Uptime monitoring
# BotCategory.PREVIEW, # Link previews (Slack, Discord)
# "OPENAI_CRAWLER_SEARCH", # Allow OpenAI crawler
],
),
],
)
@app.get("/")
async def index(request: Request):
decision = await aj.protect(request)
if decision.is_denied():
return JSONResponse({"error": "Bot detected"}, status_code=403)
return {"message": "Hello world"}
```
### Flask
```
from arcjet import arcjet_sync, detect_bot, is_spoofed_bot, Mode, BotCategory
aj = arcjet_sync(
key=arcjet_key,
rules=[
detect_bot(mode=Mode.LIVE, allow=[BotCategory.SEARCH_ENGINE]),
],
)
@app.route("/")
def index():
decision = aj.protect(request)
if decision.is_denied():
return jsonify(error="Bot detected"), 403
if any(is_spoofed_bot(r) for r in decision.results):
return jsonify(error="Spoofed bot"), 403
return jsonify(message="Hello world")
```
### 机器人类别
使用 [类别](https://docs.arcjet.com/bot-protection/identifying-bots#bot-categories) 或 [特定机器人标识符](https://github.com/arcjet/well-known-bots) 配置规则：
```
detect_bot(
mode=Mode.LIVE,
allow=[
BotCategory.SEARCH_ENGINE,
"OPENAI_CRAWLER_SEARCH",
],
)
```
可用类别：`ACADEMIC`、`ADVERTISING`、`AI`、`AMAZON`、`ARCHIVE`、`BOTNET`、`FEEDFETCHER`、`GOOGLE`、`META`、`MICROSOFT`、`MONITOR`、`OPTIMIZER`、`PREVIEW`、`PROGRAMMATIC`、`SEARCH_ENGINE`、`SLACK`、`SOCIAL`、`TOOL`、`UNKNOWN`、`VERCEL`、`YAHOO`。在 Python 中使用 `BotCategory.
` 或直接传递字符串。您还可以允许或拒绝 [特定机器人](https://arcjet.com/bot-list)。
如果指定了允许列表，则拒绝所有其他机器人。空允许列表会阻止所有机器人。拒绝列表则相反。
### 验证机器人与伪造机器人
会验证声称是知名爬虫（如 Googlebot）的机器人是否来自其已知的 IP 范围。使用 `is_spoofed_bot()` 进行检查：
```
from arcjet import is_spoofed_bot
if any(is_spoofed_bot(r) for r in decision.results):
return jsonify(error="Spoofed bot"), 403
```
有关更多详细信息，请参阅 [机器人保护文档](https://docs.arcjet.com/bot-protection)。
## 速率限制
按 IP、用户或任何自定义特征限制请求速率。Arcjet 支持令牌桶、固定窗口和滑动窗口算法。令牌桶非常适合控制 AI 令牌预算 — 设置 `capacity` 为用户可花费的最大令牌数，`refill_rate` 为每 `interval` 恢复的令牌数，并通过 `protect()` 中的 `requested` 扣除每请求令牌数。`interval` 以秒为数值。使用 `characteristics` 按用户而非按 IP 跟踪限制。
### 令牌桶（推荐用于 AI）
默认按 IP 地址跟踪速率限制。要按用户跟踪，请在规则中声明键名，然后在 `protect()` 中传递实际值：
```
from arcjet import arcjet, token_bucket, Mode
aj = arcjet(
key=arcjet_key,
rules=[
token_bucket(
characteristics=["userId"], # or ["ip.src"] for IP-based
mode=Mode.LIVE,
refill_rate=100, # tokens added per interval
interval=60, # interval in seconds
capacity=1000, # maximum tokens per bucket
),
],
)
@app.post("/chat")
async def chat(request: Request):
decision = await aj.protect(
request,
requested=5, # tokens consumed by this request
characteristics={"userId": "user_123"},
)
if decision.is_denied():
return JSONResponse({"error": "Rate limited"}, status_code=429)
```
### 固定窗口
```
from arcjet import arcjet, fixed_window, Mode
aj = arcjet(
key=arcjet_key,
rules=[
fixed_window(mode=Mode.LIVE, window=60, max=100),
],
)
```
### 滑动窗口
```
from arcjet import arcjet, sliding_window, Mode
aj = arcjet(
key=arcjet_key,
rules=[
sliding_window(mode=Mode.LIVE, interval=60, max=100),
],
)
```
有关更多详细信息，请参阅 [速率限制文档](https://docs.arcjet.com/rate-limiting)。
## 敏感信息检测
在请求到达您的 LLM 或数据存储之前检测并阻止 PII。使用内置实体类型：`EMAIL`、`PHONE_NUMBER`、`IP_ADDRESS`、`CREDIT_CARD_NUMBER`。您还可以提供自定义 `detect` 回调以处理额外模式。
```
from arcjet import arcjet, detect_sensitive_info, SensitiveInfoEntityType, Mode
aj = arcjet(
key=arcjet_key,
rules=[
detect_sensitive_info(
mode=Mode.LIVE,
deny=[
SensitiveInfoEntityType.EMAIL,
SensitiveInfoEntityType.CREDIT_CARD_NUMBER,
],
),
],
)
# 每次调用 protect() 时传递要扫描的内容
decision = await aj.protect(request, sensitive_info_value="User input to scan")
```
您可以使用自定义 `detect` 回调补充内置检测器：
```
def my_detect(tokens: list[str]) -> list[str | None]:
return ["CUSTOM_PII" if "secret" in t.lower() else None for t in tokens]
rules = [
detect_sensitive_info(
mode=Mode.LIVE,
deny=["CUSTOM_PII"],
detect=my_detect,
),
]
```
有关更多详细信息，请参阅 [敏感信息文档](https://docs.arcjet.com/sensitive-info)。
## 护盾 WAF
保护免受包括 SQL 注入、XSS、路径遍历和其他 OWASP 10 威胁在内的常见 Web 攻击。无需额外配置 — 护盾会自动分析请求模式。
```
from arcjet import arcjet, shield, Mode
aj = arcjet(
key=arcjet_key,
rules=[
shield(mode=Mode.LIVE),
],
)
```
有关更多详细信息，请参阅 [护盾文档](https://docs.arcjet.com/shield)。
## 电子邮件验证
阻止用户使用一次性、无效或无法投递的电子邮件地址注册。拒绝类型：`DISPOSABLE`、`FREE`、`INVALID`、`NO_MX_RECORDS`、`NO_GRAVATAR`。
```
from arcjet import arcjet, validate_email, EmailType, Mode
aj = arcjet(
key=arcjet_key,
rules=[
validate_email(
mode=Mode.LIVE,
deny=[
EmailType.DISPOSABLE,
EmailType.INVALID,
EmailType.NO_MX_RECORDS,
],
),
],
)
# 每次调用 protect() 时传递电子邮件
decision = await aj.protect(request, email="user@example.com")
```
有关更多详细信息，请参阅 [电子邮件验证文档](https://docs.arcjet.com/email-validation)。
## 请求过滤器
使用基于表达式的规则针对请求属性（IP 地址、标头、路径、HTTP 方法和自定义本地字段）过滤请求。
### 按国家/地区阻止
限制对特定国家/地区的访问 — 对于许可、合规或区域发布非常有用。`allow` 列表会拒绝未列出的所有国家/地区：
```
from arcjet import arcjet, filter_request, Mode
aj = arcjet(
key=arcjet_key,
rules=[
# Allow only US traffic — all other countries are denied
filter_request(
mode=Mode.LIVE,
allow=['ip.src.country eq "US"'],
),
],
)
@app.get("/")
async def index(request: Request):
decision = await aj.protect(request)
if decision.is_denied():
return JSONResponse({"error": "Access restricted in your region"}, status_code=403)
```
要限制到特定州或省份，请结合国家和区域：
```
filter_request(
mode=Mode.LIVE,
# Allow only California — useful for state-level compliance e.g. CCPA testing
allow=['ip.src.country eq "US" and ip.src.region eq "California"'],
)
```
### 阻止 VPN 和代理流量
阻止匿名流量访问敏感端点 — 对于欺诈预防、强制地理限制和减少滥用非常有用：
```
from arcjet import arcjet, filter_request, Mode
aj = arcjet(
key=arcjet_key,
rules=[
filter_request(
mode=Mode.LIVE,
deny=[
"ip.src.vpn", # VPN services
"ip.src.proxy", # Open proxies
"ip.src.tor", # Tor exit nodes
],
),
],
)
```
对于允许部分匿名流量的情况（例如 Apple Private Relay），请在调用 `protect()` 后使用 `decision.ip` 辅助方法：
```
decision = await aj.protect(request)
if decision.ip.is_vpn() or decision.ip.is_tor():
return JSONResponse({"error": "VPN traffic not allowed"}, status_code=403)
ip = decision.ip_details
if ip and ip.is_relay:
# Privacy relay (e.g. Apple Private Relay) — lower risk than a VPN
pass # allow through with custom handling
```
### 自定义本地字段
传递应用程序中的任意值以在过滤表达式中使用：
```
decision = await aj.protect(
request,
filter_local={"userId": current_user.id, "plan": current_user.plan},
)
```
这些值随后在表达式中作为 `local.userId` 和 `local.plan` 可用：
```
filter_request(
mode=Mode.LIVE,
deny=['local.plan eq "free" and ip.src.country ne "US"'],
)
```
有关更多详细信息，请参阅 [请求过滤器文档](https://docs.arcjet.com/filters)、
[IP 地理定位蓝图](https://docs.arcjet.com/blueprints/ip-geolocation)
和 [VPN/代理检测蓝图](https://docs.arcjet.com/blueprints/vpn-proxy-detection)
## IP 分析
每个决策都会返回 IP 元数据 — 无需额外 API 调用。
```
# 高级助手
if decision.ip.is_hosting():
# likely a cloud/hosting provider — often suspicious for bots
return JSONResponse({"error": "Hosting IP blocked"}, status_code=403)
if decision.ip.is_vpn() or decision.ip.is_proxy() or decision.ip.is_tor():
# apply your policy for anonymized traffic
pass
# 类型化字段访问
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
```
可用字段包括地理位置（`latitude`、`longitude`、`city`、`region`、`country`、`continent`）、网络（`asn`、`asn_name`、`asn_domain`、`asn_type`、`asn_country`）和信誉（`is_vpn`、`is_proxy`、`is_tor`、`is_hosting`、`is_relay`）。
## LangChain 示例
Arcjet 与任何 Python 代码兼容，包括 LangChain 代理和链。在此示例中，我们使用 Arcjet 保护 LangChain 代理的聊天端点，以防止提示注入、阻止机器人、防止敏感数据泄露，并在调用代理之前强制执行令牌预算。
### FastAPI + LangChain
```
from arcjet import arcjet, detect_bot, detect_prompt_injection, detect_sensitive_info, token_bucket, Mode, SensitiveInfoEntityType
aj = arcjet(
key=arcjet_key,
rules=[
detect_prompt_injection(mode=Mode.LIVE),
detect_sensitive_info(
mode=Mode.LIVE,
deny=[
SensitiveInfoEntityType.EMAIL,
SensitiveInfoEntityType.CREDIT_CARD_NUMBER,
SensitiveInfoEntityType.PHONE_NUMBER,
],
),
detect_bot(mode=Mode.LIVE, allow=["CURL"]),
token_bucket(characteristics=["userId"], mode=Mode.LIVE, refill_rate=5, interval=10, capacity=10),
],
)
@app.post("/chat")
async def chat(request: Request, body: ChatRequest):
decision = await aj.protect(
request,
requested=5,
characteristics={"userId": "user_123"},
detect_prompt_injection_message=body.message, # scan for prompt injection
sensitive_info_value=body.message, # scan for PII before sending to LLM
)
if decision.is_denied():
status = 429 if decision.reason_v2.type == "RATE_LIMIT" else 403
return JSONResponse({"error": "Denied"}, status_code=status)
reply = await chain.ainvoke({"message": body.message})
return {"reply": reply}
```
### Flask + LangChain
```
from arcjet import arcjet_sync, detect_bot, detect_prompt_injection, detect_sensitive_info, token_bucket, Mode, SensitiveInfoEntityType
aj = arcjet_sync(
key=arcjet_key,
rules=[
detect_prompt_injection(mode=Mode.LIVE),
detect_sensitive_info(
mode=Mode.LIVE,
deny=[
SensitiveInfoEntityType.EMAIL,
SensitiveInfoEntityType.CREDIT_CARD_NUMBER,
SensitiveInfoEntityType.PHONE_NUMBER,
],
),
detect_bot(mode=Mode.LIVE, allow=["CURL"]),
token_bucket(characteristics=["userId"], mode=Mode.LIVE, refill_rate=5, interval=10, capacity=10),
],
)
@app.post("/chat")
def chat():
body = request.get_json()
message = body.get("message", "") if body else ""
decision = aj.protect(
request,
requested=5,
characteristics={"userId": "user_123"},
detect_prompt_injection_message=message, # scan for prompt injection
sensitive_info_value=message, # scan for PII before sending to LLM
)
if decision.is_denied():
status = 429 if decision.reason_v2.type == "RATE_LIMIT" else 403
return jsonify(error="Denied"), status
reply = chain.invoke({"message": message})
return jsonify(reply=reply)
```
## Arcjet Guard
`arcjet.guard` 是一个更低层的 API，专为 AI 代理工具调用和后台任务设计，其中不存在 HTTP 请求对象。它为您提供细粒度的每次调用控制，包括速率限制、提示注入检测、敏感信息检测和自定义规则。
### 与 `arcjet` / `arcjet_sync` 的区别
| | `arcjet` / `arcjet_sync` | `arcjet.guard` |
| --- | --- | --- |
| **设计用途** | HTTP 请求保护 | AI 代理工具调用、后台作业 |
| **请求对象** | 需要 (`protect(request, ...)`) | 不需要 |
| **规则绑定** | 规则配置一次，通过 `protect()` kwargs 输入 | 规则配置为类，每次调用时传入输入 |
| **速率限制键** | IP 或 `characteristics` 字典 | 明确的 `key` 字符串（SHA-256 哈希后发送） |
| **自定义规则** | 不支持 | `LocalCustomRule`，带有类型化的配置/输入/数据 |
### 安装
`arcjet.guard` 已包含在 `arcjet` 包中 — 无需额外安装。
### 快速开始
```
import os
from arcjet.guard import (
launch_arcjet, # async — use launch_arcjet_sync for sync frameworks
TokenBucket,
DetectPromptInjection,
LocalDetectSensitiveInfo,
)
arcjet_key = os.getenv("ARCJET_KEY")
if not arcjet_key:
raise RuntimeError("ARCJET_KEY is required")
# 创建单个 Guard 客户端并复用。
aj = launch_arcjet(key=arcjet_key)
# 在启动时配置规则一次
user_limit = TokenBucket(
refill_rate=100,
interval_seconds=60,
max_tokens=1000,
)
prompt_scan = DetectPromptInjection()
sensitive = LocalDetectSensitiveInfo(deny=["EMAIL", "CREDIT_CARD_NUMBER"])
# 在调用时，绑定输入并应用 Guard
async def handle_tool_call(user_id: str, message: str):
decision = await aj.guard(
label="tools.weather",
rules=[
user_limit(key=user_id, requested=5),
prompt_scan(message),
sensitive(message),
],
)
if decision.conclusion == "DENY":
raise RuntimeError(f"Blocked: {decision.reason}")
# safe to proceed
```
### 同步用法
对于 Flask、Django 或其他同步框架，请使用 `launch_arcjet_sync`：
```
from arcjet.guard import launch_arcjet_sync, TokenBucket
aj = launch_arcjet_sync(key=arcjet_key)
user_limit = TokenBucket(refill_rate=10, interval_seconds=60, max_tokens=100)
def handle_tool_call(user_id: str):
decision = aj.guard(
label="tools.weather",
rules=[user_limit(key=user_id)],
)
if decision.conclusion == "DENY":
raise RuntimeError("Rate limited")
```
### 速率限制
令牌桶、固定窗口和滑动窗口算法均可用。
配置规则一次，然后针对每次调用使用 `key`（以及可选的 `requested` 令牌数）：
#### 令牌桶
```
from arcjet.guard import TokenBucket
user_limit = TokenBucket(
refill_rate=100, # tokens added per interval
interval_seconds=60, # seconds between refills
max_tokens=1000, # maximum bucket capacity
)
# 在调用时：
decision = await aj.guard(
label="tools.weather",
rules=[user_limit(key=user_id, requested=5)],
)
```
#### 固定窗口
```
from arcjet.guard import FixedWindow
team_limit = FixedWindow(
max_requests=1000,
window_seconds=3600,
)
decision = await aj.guard(
label="api.search",
rules=[team_limit(key=team_id)],
)
```
#### 滑动窗口
```
from arcjet.guard import SlidingWindow
api_limit = SlidingWindow(
max_requests=500,
interval_seconds=60,
)
decision = await aj.guard(
label="api.query",
rules=[api_limit(key=user_id)],
)
```
### 提示注入检测
```
from arcjet.guard import DetectPromptInjection
prompt_scan = DetectPromptInjection()
decision = await aj.guard(
label="tools.weather",
rules=[prompt_scan(user_message)],
)
if decision.conclusion == "DENY":
print("Prompt injection detected")
```
### 敏感信息检测
在本地通过 WASM 检测 PII — 原始文本不会离开 SDK。内置实体类型：`EMAIL`、`PHONE_NUMBER`、`IP_ADDRESS`、`CREDIT_CARD_NUMBER`。
```
from arcjet.guard import LocalDetectSensitiveInfo
sensitive = LocalDetectSensitiveInfo(
deny=["EMAIL", "CREDIT_CARD_NUMBER"],
)
decision = await aj.guard(
label="tools.send_email",
rules=[sensitive(user_input)],
)
```
### 自定义规则
定义类型化的自定义规则，继承 `LocalCustomRule` 并重写 `evaluate`（同步）或 `evaluate_async`（异步）：
```
from typing import TypedDict
from arcjet.guard import LocalCustomRule, CustomEvaluateResult
class TopicConfig(TypedDict):
blocked_topic: str
class TopicInput(TypedDict):
topic: str
class TopicData(TypedDict):
matched: str
class TopicBlockRule(LocalCustomRule[TopicConfig, TopicInput, TopicData]):
def evaluate(
self,
config: TopicConfig,
input: TopicInput,
) -> CustomEvaluateResult:
if input["topic"] == config["blocked_topic"]:
return CustomEvaluateResult(
conclusion="DENY",
data={"matched": input["topic"]},
)
return CustomEvaluateResult(conclusion="ALLOW")
rule = TopicBlockRule(config={"blocked_topic": "weapons"})
inp = rule(data={"topic": user_topic})
decision = await aj.guard(rules=[inp], label="content")
# 访问类型化结果
r = inp.result(decision)
if r and r.conclusion == "DENY":
print(f"Blocked topic: {r.data['matched']}")
```
### 每个规则的结果
配置的规则和绑定的输入均提供类型化的结果访问器：
```
user_limit = TokenBucket(refill_rate=10, interval_seconds=60, max_tokens=100)
inp = user_limit(key=user_id, requested=5)
decision = await aj.guard(label="tools.weather", rules=[inp])
# 来自绑定的输入（匹配精确调用）
r = inp.result(decision)
if r:
print(r.remaining_tokens, r.max_tokens)
# 来自配置的规则（匹配此规则的所有调用）
r = user_limit.result(decision)
# 仅检查拒绝的结果
denied = inp.denied_result(decision)
if denied:
print(f"Rate limited — resets at {denied.reset_at_unix_seconds}")
```
### 决策 API
```
decision = await aj.guard(label="tools.weather", rules=[...])
# 第一层：结论与原因
decision.conclusion # "ALLOW" or "DENY"
decision.reason # "RATE_LIMIT", "PROMPT_INJECTION", "SENSITIVE_INFO", "CUSTOM", "ERROR", etc.
# 第二层：错误检测
decision.has_error() # True if any rule errored or the server reported diagnostics
# 第三层：每规则结果（参见上方的“每规则结果”）
for result in decision.results:
print(result.type, result.conclusion)
```
### `guard()` 参数参考
| 参数 | 类型 | 描述 |
| --- | --- | --- |
| `rules` | `Sequence[RuleWithInput]` | 绑定的规则输入（必需） |
| `label` | `str` | 标识此 guard 调用的标签（必需） |
| `metadata` | `dict[str, str] \| None` | 可选键值元数据 |
### DRY_RUN 模式
所有 guard 规则都接受 `mode` 参数。使用 `"DRY_RUN"` 来评估规则而不阻止：
```
user_limit = TokenBucket(
refill_rate=10,
interval_seconds=60,
max_tokens=100,
mode="DRY_RUN",
)
```
## 最佳实践
### 单例模式
在启动时创建一个 Arcjet 客户端并在所有请求中重复使用：
```
# 良好 —— 一个实例，启动时创建一次
aj = arcjet(key=arcjet_key, rules=[...])
# 糟糕 —— 每个请求创建新实例会浪费资源
@app.get("/")
async def index(request: Request):
aj = arcjet(key=arcjet_key, rules=[...]) # don't do this
```
### DRY_RUN 模式用于测试
使用 `Mode.DRY_RUN` 测试规则而不阻塞流量。决策会被记录但请求会被允许通过：
```
aj = arcjet(
key=arcjet_key,
rules=[
detect_bot(mode=Mode.DRY_RUN, allow=[]),
token_bucket(mode=Mode.DRY_RUN, refill_rate=5, interval=10, capacity=10),
],
)
```
### 代理配置
当在负载均衡器或反向代理后面运行时，配置可信 IP 以便 Arcjet 从 `X-Forwarded-For` 解析真实的客户端 IP：
```
aj = arcjet(
key=arcjet_key,
rules=[...],
proxies=["10.0.0.0/8", "192.168.0.1"],
)
```
### 异步与同步客户端
对于 FastAPI 等异步框架使用 `arcjet`（异步）。对于 Flask 等同步框架使用 `arcjet_sync`：
```
from arcjet import arcjet, arcjet_sync
# 异步 —— 适用于 FastAPI、Starlette 等
aj_async = arcjet(key=arcjet_key, rules=[...])
decision = await aj_async.protect(request)
# 同步 —— 适用于 Flask、Django 等
aj_sync = arcjet_sync(key=arcjet_key, rules=[...])
decision = aj_sync.protect(request)
```
### `protect()` 参数参考
所有参数都是可选的关键词参数，与 `request` 一起传递：
| 参数 | 类型 | 用途 |
| --- | --- | --- |
| `requested` | `int` | 令牌桶速率限制 |
| `characteristics` | `dict[str, Any]` | 速率限制（传递规则配置中声明的键对应的值） |
| `detect_prompt_injection_message` | `str` | 提示注入检测 |
| `sensitive_info_value` | `str` | 敏感信息检测 |
| `email` | `str` | 电子邮件验证 |
| `filter_local` | `dict[str, str]` | 请求过滤器（`local.*` 字段） |
| `ip_src` | `string` | 高级手动 IP 覆盖 |
### 决策响应
```
decision = await aj.protect(request)
# 顶层检查
decision.is_denied() # True if any rule denied the request
decision.is_allowed() # True if all rules allowed the request
decision.is_error() # True if Arcjet encountered an error (fails open)
# reason_v2.type 值："BOT"、"RATE_LIMIT"、"SHIELD"、"EMAIL"、"ERROR"、"FILTER"
if decision.reason_v2.type == "RATE_LIMIT":
print(decision.reason_v2.remaining) # tokens/requests remaining
elif decision.reason_v2.type == "BOT":
print(decision.reason_v2.denied) # list of denied bot names
print(decision.reason_v2.spoofed) # list of spoofed bot names
# 每规则结果（用于精细处理）
for result in decision.results:
print(result.reason_v2.type, result.is_denied())
```
### 错误处理
Arcjet 设计为失败开放 — 如果服务不可用，请求会被允许通过。如果您的用例需要明确检查错误，请显式检查：
```
decision = await aj.protect(request)
if decision.is_error():
# Arcjet service error — fail open or apply fallback policy
pass
elif decision.is_denied():
return JSONResponse({"error": "Denied"}, status_code=403)
```
## 支持
本仓库遵循 [Arcjet 支持策略](https://docs.arcjet.com/support)。
## 安全性
本仓库遵循 [Arcjet 安全策略](https://docs.arcjet.com/security)。
## 兼容性
本仓库维护的包兼容 Python 3.10 及以上版本。
## 许可证
根据 [Apache License, Version 2.0](http://www.apache.org/licenses/LICENSE-2.0) 授权。 
