Skip to content

LangChain agent guard

LangChain tools and create_agent agents call authored tools. Arcjet Guard sits at those boundaries so a policy can allow or deny the action before a side effect runs.

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.

Use protect() on HTTP routes. Use the helpers on this page for agent tools and other actions that have no HTTP request.

You need an Arcjet account and an ARCJET_KEY. Launch one client at module scope and reuse it.

Framework wrappers take an action string such as email.sent. That slug selects the matching remote policy and names the event in the Arcjet Console. Direct guard() calls use the field name label for the same slug. Don’t pass label to a wrapper such as guardTool().

You can submit SDK rules in code, rely on a published remote policy, or combine both. For more information about the decision model, see Agent guards.

Vercel AI SDK, LangGraph JS, Eve, Mastra, and Claude Agent SDK wrappers are on Framework integrations.

This is the Python SDK. It is not the JavaScript LangGraph Graph API adapter (StateGraph + ToolNode).

guard_action and guard_action_sync ship in the core arcjet.guard package, so they need no extra.

Install an extra only for the LangChain surfaces you use:

Terminal window
pip install "arcjet[langchain]"

arcjet[langchain] provides guard_tool, ArcjetCaptureHandler, and ArcjetAsyncCaptureHandler. It depends on langchain-core>=1.2.5,<2.

Terminal window
pip install "arcjet[langchain-agents]"

arcjet[langchain-agents] adds ArcjetMiddleware and ToolPolicy. It depends on langchain>=1.3,<2 and langgraph>=1.2,<2.

Import LangChain helpers from one path:

from arcjet.guard.langchain import (
ArcjetCaptureHandler,
ArcjetMiddleware,
ToolPolicy,
guard_tool,
)

Importing arcjet.guard.langchain does not load LangGraph. ArcjetMiddleware and ToolPolicy raise ImportError if you haven’t installed arcjet[langchain-agents].

Launch one client at module scope:

from arcjet.guard import launch_arcjet
arcjet = launch_arcjet(key=ARCJET_KEY)

Use launch_arcjet_sync with Flask, Django, or other sync code.

Pick the surface that matches what you hold when the effect runs:

You haveUseNeedsBlocks a call?
Any Python callableguard_action / guard_action_syncarcjetYes
A LangChain BaseTool you call yourselfguard_toolarcjet[langchain]Yes
An agent from create_agentArcjetMiddleware + ToolPolicyarcjet[langchain-agents]Yes
A chain or agent you want to observeArcjetCaptureHandler / ArcjetAsyncCaptureHandlerarcjet[langchain]No

If you can name the tool at wiring time, guard_tool is the smaller change. It returns a drop-in replacement for the tool, so nothing downstream changes. If the model picks the tool and you want one policy per tool name, use the middleware. They compose: a guarded tool called from a guarded agent evaluates once per policy.

guard_tool, guard_action, and ToolPolicy share this contract:

OptionRequiredDescription
actionYesGuard label and capture name. Use resource.verb in the past tense, such as email.sent.
rulesNoBound SDK rule inputs. Empty is normal and still contacts Guard so a remote policy can apply.
actorNoTrusted identity. On guard_tool, a callable receives RunnableConfig. On ToolPolicy, a callable receives parsed arguments.
inputsNoNamed values built with server_input.* or local_input.*. On guard_tool, a callable receives (arguments, config). On ToolPolicy, a callable receives parsed arguments.
on_guard_errorNo"deny" (default) or "allow". A real DENY always blocks.
metadataNoNested JSON on ToolPolicy.

ArcjetMiddleware also accepts guard (optional if you called register_arcjet()), policies (tool name to ToolPolicy), and tools (the same sequence you passed to create_agent, so a typo is refused at init).

guard_action wraps a no-argument async callable. guard_action_sync wraps a no-argument sync callable. Neither extra is required.

On DENY the callable does not run and the helper raises ArcjetDeniedError. If Guard cannot be evaluated, the default on_guard_error="deny" raises ArcjetUnavailableError.

from arcjet.guard import (
DetectPromptInjection,
TokenBucket,
guard_action,
launch_arcjet,
)
arcjet = launch_arcjet(key=ARCJET_KEY)
job_limit = TokenBucket(
refill_rate=10,
interval_seconds=60,
max_tokens=10,
bucket="jobs",
)
inbound = DetectPromptInjection()
async def process_job(user_id: str, message: str) -> str:
return await guard_action(
lambda: run_job(message),
guard=arcjet,
action="job.processed",
rules=[
job_limit(key=user_id, requested=1),
inbound(message),
],
)

guard_tool wraps a synchronous or asynchronous BaseTool. Use it when your code calls the tool. Use the async Guard client with ainvoke() and the sync client with invoke().

On DENY the wrapped tool does not run. The helper raises ArcjetToolDeniedError, or follows the tool’s handle_tool_error behavior when you configured that on the tool before wrapping. If Guard cannot be evaluated, the default raises ArcjetToolUnavailableError.

Configure and narrow the tool before you call guard_tool(). Changes to args_schema, handle_tool_error, callbacks, or response_format after wrapping do not reach the wrapped tool.

from arcjet.guard import TokenBucket, launch_arcjet
from arcjet.guard.langchain import guard_tool
from langchain_core.tools import tool
arcjet = launch_arcjet(key=ARCJET_KEY)
lookup_limit = TokenBucket(
refill_rate=10,
interval_seconds=60,
max_tokens=10,
bucket="lookups",
)
@tool
async def lookup_order(order_id: str, note: str) -> dict:
"""Look up an order by ID."""
return {"order_id": order_id, "note": note, "status": "shipped"}
guarded_lookup_order = guard_tool(
guard=arcjet,
tool=lookup_order,
action="order.looked-up",
rules=[lookup_limit(key="orders", requested=1)],
)

To hide an argument from the model, narrow args_schema on the tool first, then wrap it:

from pydantic import BaseModel, ConfigDict
class PublicEmailArgs(BaseModel):
model_config = ConfigDict(extra="forbid")
to: str
send_email.args_schema = PublicEmailArgs
guarded_send_email = guard_tool(
guard=arcjet,
tool=send_email,
action="email.sent",
)

When the model chooses the tool, pass ArcjetMiddleware to create_agent. Match each consequential tool name to a ToolPolicy. Tools with no policy pass through unguarded.

Pass tools= the same sequence you gave create_agent. The middleware matches each policy by tool name, so a typo or a renamed @tool function leaves that tool unguarded. When you pass the tools, the middleware rejects a policy key that names none of them.

The client is optional. Without guard=, the checkpoint uses the client registered with register_arcjet().

On DENY the tool does not run and the middleware raises ArcjetDeniedError. If Guard cannot be evaluated, the default raises ArcjetUnavailableError.

from langchain.agents import create_agent
from arcjet.guard import TokenBucket, launch_arcjet
from arcjet.guard.langchain import ArcjetMiddleware, ToolPolicy
arcjet = launch_arcjet(key=ARCJET_KEY)
send_limit = TokenBucket(
refill_rate=5,
interval_seconds=60,
max_tokens=5,
bucket="email",
)
agent = create_agent(
model="openai:gpt-4o",
tools=[send_email, search_orders],
middleware=[
ArcjetMiddleware(
guard=arcjet,
policies={
"send_email": ToolPolicy(
action="email.sent",
rules=[send_limit(key="email", requested=1)],
)
},
tools=[send_email, search_orders],
)
],
)

ArcjetCaptureHandler and ArcjetAsyncCaptureHandler record chain, model, and tool lifecycle events. They cannot deny a call. LangChain ignores callback return values.

Use a handler only for visibility. Put enforcement on guard_tool or ArcjetMiddleware.

from arcjet.guard.langchain import ArcjetCaptureHandler
await agent.ainvoke(
{"messages": [{"role": "user", "content": prompt}]},
config={"callbacks": [ArcjetCaptureHandler()]},
)

Checkpoint surfaces default to on_guard_error="deny":

  • guard_action / guard_action_sync
  • guard_tool
  • ArcjetMiddleware

If Guard cannot be evaluated, the action does not run. Set on_guard_error="allow" only when you can accept running the action without a complete security decision. The helpers reject any other value.

A DENY conclusion always blocks, regardless of on_guard_error. The errors are distinct: a denial means policy evaluated and said no (ArcjetDeniedError / ArcjetToolDeniedError). Unavailability means the check never happened (ArcjetUnavailableError / ArcjetToolUnavailableError).

The core guard() call still fails open. It returns ALLOW, and has_failed_open() returns True. The wrappers that sit around an effect fail closed.

Pass one correlation ID into ainvoke() so middleware and guarded tools share one Sequence. Don’t mint a new ID per turn.

await agent.ainvoke(
{"messages": [{"role": "user", "content": prompt}]},
config={"configurable": {"arcjet_correlation_id": session_id}},
)

The helpers read configurable.arcjet_correlation_id, then metadata.arcjet_correlation_id, then an enclosing arcjet_sequence. They do not use LangChain’s run_id.

from arcjet.guard import arcjet_sequence
with arcjet_sequence(correlation_id=session_id):
await agent.ainvoke({"messages": [{"role": "user", "content": prompt}]})

Derive the ID from a session the caller already has. A generated ID still joins this run’s events, but it builds a Sequence that nobody searches for.

  • Don’t use ArcjetCaptureHandler or ArcjetAsyncCaptureHandler to deny a call. Callbacks cannot block.
  • Don’t change args_schema, handle_tool_error, callbacks, or response_format on a tool after guard_tool().
  • Don’t import ArcjetMiddleware or ToolPolicy unless you installed arcjet[langchain-agents].
  • Don’t use these helpers with the JavaScript LangGraph Graph API adapter.
  • Don’t mint a new arcjet_correlation_id per turn.
  • Callable with no LangChain extra: guard_action around a no-argument async function.
  • Tool you invoke yourself: guard_tool with action and rules.
  • Model-selected tools: ArcjetMiddleware plus a ToolPolicy per consequential tool name.
  • Observe only: ArcjetCaptureHandler on config.callbacks. This cannot deny a call.