Arcjet Guards

Guards apply Arcjet security rules inside AI agent tool calls and anywhere else you process untrusted input. Pass inputs directly, get a decision back. No Request object needed.

They are for the places HTTP middleware can’t reach: inside tool handlers, between agent steps, in queue consumers, and across agentic pipelines that fan out across many services.

Stop hidden instructions from hijacking your agent

A webpage your agent fetched contains hidden instructions telling it to email your customer list to an attacker. Guards detects the prompt injection before that content re-enters the model’s context.

Prevent runaway spend

A runaway agent loop burns $8,000 in model spend overnight. Guards enforces per-user token budgets inside the loop itself, denying calls once the bucket is empty.

Keep PII off third-party models

A background job ships support tickets to a third-party LLM for summarization. Guards blocks personally identifiable information (emails, card numbers, etc.) before the message leaves your system.

Guard vs. Protect

Arcjet has two entrypoints. Pick the one that matches the surface you need to secure:

	Protect (request SDKs)	Guard (`@arcjet/guard`, `arcjet.guard`)
Designed for	HTTP route handlers, API endpoints	AI tool calls, MCP servers, queue workers, background jobs
Request object	Required (`protect(request, ...)`)	Not needed — pass inputs directly
Rule binding	Rules configured once, input via `protect()` kwargs / options	Rules configured as classes, called with input per invocation
Rate limit key	IP or `characteristics` dict	Explicit `key` string at call time
Rate Limiting	✅	✅
Prompt Injection Detection	✅	✅
Sensitive Information Detection	✅	✅
Custom Rules	—	✅
Bot Protection	✅	—
Shield WAF	✅	—
Email Validation	✅	—
Request Filters	✅	—
IP Analysis	✅	—

A single application can use both — request-based protection on API routes and guards inside tool handlers / queue workers. Both share the same Arcjet account and dashboard.

When to use Guards

Use a guard wherever an Arcjet decision is needed but no HTTP request exists:

AI agent tool calls — inside each tool/skill function so denied actions never reach the LLM call site.
MCP servers — stdio or SSE-based MCP servers don’t receive HTTP requests from the LLM client. Guard each tool handler with a hardcoded label.
Queue consumers / background jobs — protect per-job budgets and scan job payloads for prompt injection or PII before processing.
Agentic pipelines — fan-out workflows (e.g. one user prompt → many downstream tool calls) need per-step enforcement that survives the boundary between services.
Re-checking content returned to the model — guard prompt injection on tool results when the tool fetches untrusted content (web pages, third party APIs).

If the protection site already has a Request object — Express, Next.js route handlers, FastAPI handlers, Lambda HTTP events — use the request-based SDKs instead. Request-based SDKs include extra signals (IP analysis, headers, bot fingerprinting) that guards do not.

Get started

Install the Arcjet skill into your AI coding agent. It will detect your language, install the package, configure rules, and wire up guard() calls inline in your tool handlers.

Supported languages: JavaScript / TypeScript (@arcjet/guard >= 1.4.0) and Python (arcjet >= 0.7.0).

npx skills add arcjet/skills

Then describe what you want to protect — for example: “rate limit my tool calls per user and block prompt injection” or “secure my MCP server” or “protect my queue worker”.

Manual install

If you’d rather wire it up by hand:

JavaScript / TypeScript
Python

npm install @arcjet/guard

Requires Node.js / Bun / Deno and @arcjet/guard >= 1.4.0.

pip install arcjet
# or
uv add arcjet

Requires Python 3.10+ and arcjet >= 0.7.0. Guard ships in the arcjet package — there is no separate package to install.

Set ARCJET_KEY in your environment — retrieve it from the Arcjet dashboard, the Arcjet CLI, or the MCP server.

Architecture

Client at module scope

The guard client holds a persistent connection to the Arcjet decision service. Create it once at module scope and reuse it for every call:

JavaScript / TypeScript
Python

1
import { launchArcjet } from "@arcjet/guard";
2

3
const arcjet = launchArcjet({ key: process.env.ARCJET_KEY! });

import os
from arcjet.guard import launch_arcjet  # launch_arcjet_sync for sync code

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

Use launch_arcjet for async code (FastAPI handlers, agent loops with AsyncOpenAI, etc.) and launch_arcjet_sync for sync code (Celery tasks, Flask jobs).

Rules at module scope

Rate limit state is tracked server-side by the rule’s bucket name, so recreating rules per call won’t break counting. However, you should still declare rules at module scope:

The per-rule result accessors (userLimit.deniedResult(decision) in JS, user_limit.denied_result(decision) in Python) only work when you have a stable reference.
It avoids unnecessary object allocation on every invocation.
It keeps rule configuration visible and centralized.

JavaScript / TypeScript
Python

1
import { tokenBucket, detectPromptInjection } from "@arcjet/guard";
2

3
const userLimit = tokenBucket({
4
  refillRate: 10,
5
  intervalSeconds: 60,
6
  maxTokens: 100,
7
  bucket: "user-tools", // name explicitly to avoid collisions across rules
8
});
9

10
const piRule = detectPromptInjection();

from arcjet.guard import TokenBucket, DetectPromptInjection

user_limit = TokenBucket(
    refill_rate=10,
    interval_seconds=60,
    max_tokens=100,
    bucket="user-tools",  # name explicitly to avoid collisions across rules
)

pi_rule = DetectPromptInjection()

`guard()` at the operation, with a hardcoded label

Call guard() wherever you already know exactly what operation is happening — usually inside the specific tool/task function. The label should be a hardcoded string literal ("tools.get_weather", not `tools.${name}`). Hardcoded labels stay greppable and the dashboard groups by them; interpolation produces a sea of distinct-looking calls.

Pass metadata whenever you have useful auditing context (userId, requestId, etc.) — it appears in the dashboard alongside the decision and makes debugging much easier.

JavaScript / TypeScript
Python

1
async function getWeather(city: string, userId: string) {
2
  const decision = await arcjet.guard({
3
    label: "tools.get_weather",
4
    rules: [userLimit({ key: userId, requested: 1 })],
5
    metadata: { userId },
6
  });
7
  if (decision.conclusion === "DENY") {
8
    throw new Error(`Blocked: ${decision.reason}`);
9
  }
10
  // ... do the work
11
}

async def get_weather(city: str, user_id: str):
    decision = await arcjet.guard(
        label="tools.get_weather",
        rules=[user_limit(key=user_id, requested=1)],
        metadata={"user_id": user_id},
    )
    if decision.conclusion == "DENY":
        raise RuntimeError(f"Blocked: {decision.reason}")
    # ... do the work

Rules

Every guard rule accepts mode ("LIVE" enforces, "DRY_RUN" logs only), label (an observability label that appears in the dashboard), and metadata (arbitrary key-value context).

Rate limiting

Three algorithms are available — pick the one that matches the cost model of the operation you’re protecting:

Token bucket — variable cost per call (e.g. LLM token spend). Bursts up to maxTokens, refills at refillRate per intervalSeconds.
Fixed window — hard cap per window (e.g. 1000 tool calls per hour). Resets at window boundaries.
Sliding window — smooth limit without boundary bursts (e.g. 500 requests per rolling 60 seconds).

All three require a key and a bucket name at call time. Without a key, limits would be global across all callers.

JavaScript / TypeScript
Python

1
import { tokenBucket, fixedWindow, slidingWindow } from "@arcjet/guard";
2

3
const userLimit = tokenBucket({
4
  refillRate: 100,
5
  intervalSeconds: 60,
6
  maxTokens: 1000,
7
  bucket: "user-tools",
8
});
9

10
const teamLimit = fixedWindow({
11
  maxRequests: 1000,
12
  windowSeconds: 3600,
13
  bucket: "team-api",
14
});
15

16
const apiLimit = slidingWindow({
17
  maxRequests: 500,
18
  intervalSeconds: 60,
19
  bucket: "public-api",
20
});
21

22
// At call time:
23
const decision = await arcjet.guard({
24
  label: "tools.weather",
25
  rules: [userLimit({ key: userId, requested: 5 })],
26
});

from arcjet.guard import TokenBucket, FixedWindow, SlidingWindow

user_limit = TokenBucket(
    refill_rate=100,
    interval_seconds=60,
    max_tokens=1000,
    bucket="user-tools",
)

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

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

# At call time:
decision = await arcjet.guard(
    label="tools.weather",
    rules=[user_limit(key=user_id, requested=5)],
)

Picking a key when there’s no user: some call sites have no per-user context — a stdio MCP server where the client is the only caller, or a single-tenant worker. Don’t fake it with an empty string. Use an identifier that matches the scope of the limit (deployment name, hostname, the MCP session id if exposed) and add a comment if it’s deliberately global.

Prompt injection detection

Use on any untrusted text before it reaches a model or is used as a tool argument. Also useful on tool call results when the tool fetches content from untrusted sources (web pages, third-party APIs).

JavaScript / TypeScript
Python

1
import { detectPromptInjection } from "@arcjet/guard";
2

3
const piRule = detectPromptInjection();
4

5
const decision = await arcjet.guard({
6
  label: "tools.weather",
7
  rules: [piRule(userMessage)],
8
});

from arcjet.guard import DetectPromptInjection

pi_rule = DetectPromptInjection()

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

Sensitive information detection

Detects PII (email addresses, phone numbers, IP addresses, credit card numbers) in text locally via WASM — the raw text never leaves the SDK, which matters for compliance. Built-in entity types: EMAIL, PHONE_NUMBER, IP_ADDRESS, CREDIT_CARD_NUMBER.

JavaScript / TypeScript
Python

1
import { localDetectSensitiveInfo } from "@arcjet/guard";
2

3
const sensitive = localDetectSensitiveInfo({
4
  deny: ["EMAIL", "CREDIT_CARD_NUMBER"],
5
});
6

7
const decision = await arcjet.guard({
8
  label: "tools.send_email",
9
  rules: [sensitive(userInput)],
10
});

from arcjet.guard import LocalDetectSensitiveInfo

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

decision = await arcjet.guard(
    label="tools.send_email",
    rules=[sensitive(user_input)],
)

allow and deny are mutually exclusive — pass only one.

Custom rules

Custom rules let you plug in arbitrary local evaluation logic with typed config, input, and result data. They run inside the SDK — no data leaves your environment.

JavaScript / TypeScript
Python

See the @arcjet/guard README for the full TypeScript API.

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, payload):
        if payload["topic"] == config["blocked_topic"]:
            return CustomEvaluateResult(
                conclusion="DENY",
                data={"matched": payload["topic"]},
            )
        return CustomEvaluateResult(conclusion="ALLOW")


rule = TopicBlockRule(config={"blocked_topic": "weapons"})
inp = rule(data={"topic": user_topic})
decision = await arcjet.guard(label="content", rules=[inp])

r = inp.result(decision)
if r and r.conclusion == "DENY":
    print(f"Blocked topic: {r.data['matched']}")

For sync code, override evaluate (as shown). For async code, override evaluate_async instead.

Decision handling

decision.conclusion is either "ALLOW" or "DENY". Always check before proceeding. For useful error messages, branch on which rule denied — not just on DENY. Each rule defined at module scope exposes typed result accessors that return rule-specific info (e.g. remaining tokens, reset time) for actionable error messages.

JavaScript / TypeScript
Python

1
if (decision.conclusion === "DENY") {
2
  const rateLimited = userLimit.deniedResult(decision);
3
  if (rateLimited) {
4
    throw new Error(
5
      `Rate limited — retry in ${rateLimited.resetInSeconds}s`,
6
    );
7
  }
8
  if (decision.reason.isPromptInjection()) {
9
    throw new Error("Input flagged as prompt injection");
10
  }
11
  throw new Error("Blocked");
12
}
13

14
// hasError() means a rule errored or the server reported diagnostics.
15
// The SDK failed open — log it but don't block the caller.
16
if (decision.hasError()) {
17
  console.warn("Arcjet rule error");
18
}

import logging
import time

logger = logging.getLogger(__name__)

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 RuntimeError(f"Rate limited — retry in {retry_in}s")
    if decision.reason == "PROMPT_INJECTION":
        raise RuntimeError("Input flagged as prompt injection")
    raise RuntimeError("Blocked")

# has_error() means a rule errored or the server reported diagnostics.
# The SDK failed open — log it but don't block the caller.
if decision.has_error():
    logger.warning("Arcjet rule error")

In Python, the guard’s decision.reason is a string literal ("RATE_LIMIT" | "PROMPT_INJECTION" | "SENSITIVE_INFO" | "CUSTOM" | "ERROR" | "NOT_RUN" | "UNKNOWN"). Rate limit denied results expose reset_at_unix_seconds, remaining_tokens, max_tokens (token bucket) or remaining_requests, max_requests (fixed and sliding window).

DRY_RUN mode

Every guard rule accepts a mode parameter. Use "DRY_RUN" to evaluate rules without blocking — decisions are logged in the Arcjet dashboard but the conclusion is always "ALLOW". This lets you tune rule configuration in production before promoting to "LIVE".

JavaScript / TypeScript
Python

1
const userLimit = tokenBucket({
2
  refillRate: 10,
3
  intervalSeconds: 60,
4
  maxTokens: 100,
5
  bucket: "user-tools",
6
  mode: "DRY_RUN",
7
});

user_limit = TokenBucket(
    refill_rate=10,
    interval_seconds=60,
    max_tokens=100,
    bucket="user-tools",
    mode="DRY_RUN",
)

Common patterns

Guarding an MCP tool

MCP servers don’t receive HTTP requests — they’re invoked by an MCP client over stdio or SSE. Use a guard at the tool handler with a hardcoded label and a key that matches the scope of your server (single-user, single-tenant, or session-based).

JavaScript / TypeScript
Python

1
server.tool("query_database", async ({ query, sessionId }) => {
2
  const decision = await arcjet.guard({
3
    label: "mcp.query_database",
4
    rules: [
5
      queryLimit({ key: sessionId ?? "default", requested: 1 }),
6
      piRule(query),
7
    ],
8
    metadata: { sessionId },
9
  });
10

11
  if (decision.conclusion === "DENY") {
12
    return { content: [{ type: "text", text: "Blocked by Arcjet" }] };
13
  }
14

15
  // ... execute the query
16
});

@mcp.tool()
async def query_database(query: str, session_id: str | None = None):
    decision = await arcjet.guard(
        label="mcp.query_database",
        rules=[
            query_limit(key=session_id or "default", requested=1),
            pi_rule(query),
        ],
        metadata={"session_id": session_id or ""},
    )

    if decision.conclusion == "DENY":
        return [{"type": "text", "text": "Blocked by Arcjet"}]

    # ... execute the query

Guarding tool results

Prompt injection can arrive through tool results — a fetch tool retrieves a page containing hidden instructions, which then re-enter the model context. Guard the result as well as the input:

JavaScript / TypeScript
Python

1
async function fetchTool({ url }: { url: string }, userId: string) {
2
  const content = await fetch(url).then((r) => r.text());
3

4
  const decision = await arcjet.guard({
5
    label: "tools.fetch",
6
    metadata: { userId },
7
    rules: [piRule(content)],
8
  });
9

10
  if (decision.conclusion === "DENY") {
11
    // Return a safe placeholder rather than the injected content
12
    return { content: "[Content blocked: prompt injection detected]" };
13
  }
14

15
  return { content };
16
}

import httpx

async def fetch_tool(url: str, user_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        content = (await client.get(url)).text

    decision = await arcjet.guard(
        label="tools.fetch",
        metadata={"user_id": user_id},
        rules=[pi_rule(content)],
    )

    if decision.conclusion == "DENY":
        # Return a safe placeholder rather than the injected content
        return {"content": "[Content blocked: prompt injection detected]"}

    return {"content": content}

Queue worker / background job

Use the sync variant of the guard client when running inside a sync framework (Celery, Flask jobs). Pass the user ID from the job payload as the rate-limit key:

JavaScript / TypeScript
Python

1
// Bull / BullMQ worker example
2
worker.process(async (job) => {
3
  const decision = await arcjet.guard({
4
    label: "queue.summarize",
5
    rules: [
6
      userTaskLimit({ key: job.data.userId, requested: 1 }),
7
      piRule(job.data.input),
8
    ],
9
    metadata: { userId: job.data.userId, jobId: job.id },
10
  });
11

12
  if (decision.conclusion === "DENY") {
13
    throw new Error("Blocked by Arcjet");
14
  }
15

16
  // ... process the job
17
});

# Celery task example — use launch_arcjet_sync for sync workers
import os

from arcjet.guard import launch_arcjet_sync, TokenBucket, DetectPromptInjection

arcjet = launch_arcjet_sync(key=os.environ["ARCJET_KEY"])
user_task_limit = TokenBucket(
    refill_rate=10,
    interval_seconds=60,
    max_tokens=100,
    bucket="queue-tasks",
)
pi_rule = DetectPromptInjection()


@celery.task
def summarize(payload: dict):
    decision = arcjet.guard(
        label="queue.summarize",
        rules=[
            user_task_limit(key=payload["user_id"], requested=1),
            pi_rule(payload["input"]),
        ],
        metadata={"user_id": payload["user_id"]},
    )

    if decision.conclusion == "DENY":
        raise RuntimeError("Blocked by Arcjet")

    # ... process the task

Best practices

Hardcode the label string. Use "tools.get_weather", not `tools.${name}`. Hardcoded labels stay greppable and the dashboard groups by them.
Pass metadata for auditing. userId, sessionId, requestId all show up alongside the decision in the dashboard.
Always set bucket on rate limit rules. Default buckets share counters across all rules of that type.
Always pass a key to rate limit rules. Without it, limits are global across all callers.
Branch on the denied result, not just DENY. Specific error messages help callers retry intelligently (e.g. “rate limited — retry in 12s” vs a generic “blocked”).
Fail open on errors. decision.hasError() (JS) / decision.has_error() (Python) means a rule errored but the SDK didn’t block. Log it; don’t deny.

Reference

Arcjet Python SDK reference: Guard
arcjet/skills — AI coding agent skills for Arcjet
arcjet/arcjet-js — JavaScript / TypeScript SDK (@arcjet/guard)
arcjet/arcjet-py — Python SDK (arcjet)