Integrate Arcjet Guard into Python LangChain

arcjet.guard.langchain wraps the agent's existing Arcjet client. It never talks to the Arcjet API itself. Shared Guard fundamentals (client, rules, labels, decisions, capture, registration) live in ../arcjet/references/guards_python.md. Load that reference for anything that is not LangChain-specific.

Four surfaces, one decision rule:

• Any Python callable → guard_action / guard_action_sync in core arcjet.guard. No LangChain extra. Details in the shared Python Guard reference.
• A LangChain BaseTool you call yourself → guard_tool. DENY raises ArcjetToolDeniedError (the tool's handle_tool_error may convert it).
• create_agent (the model chooses tools) → ArcjetMiddleware + ToolPolicy. Tools without a policy pass through.
• Observe a chain or agent → ArcjetCaptureHandler / ArcjetAsyncCaptureHandler. These cannot deny.

This is Python create_agent. The merged page https://docs.arcjet.com/guards/langchain/ covers both languages — JS is @arcjet/guard/langchain/v1 (createAgent / wrapToolCall), not this module. It is not LangGraph JS StateGraph / ToolNode (docs https://docs.arcjet.com/guards/langgraph/). Importing arcjet.guard.langchain does not load LangGraph. Referencing ArcjetMiddleware or ToolPolicy without arcjet[langchain-agents] raises and names that extra.

Screen inbound before agent.ainvoke — there is no inbound helper

There is no inbound helper. Put prompt-injection (and other inbound rules) in the application with core aj.guard(...) before ainvoke / invoke. Core guard() fails open: ALLOW is not proof the rules ran. Gate on decision.has_failed_open() when this call site must fail closed. guard_tool / ArcjetMiddleware already default to that.

Capture handlers never block

LangChain ignores what a callback returns. ArcjetCaptureHandler only records. Policy lives in guard_action / guard_tool / ArcjetMiddleware.

Configure the tool before guard_tool()

Narrow args_schema, set handle_tool_error / callbacks / response_format on the tool you still hold, then wrap. Changes on the guarded handle do not reach the call.

Questions to ask the human first

Ask only what you cannot infer from the code; suggest defaults.

1. Which tools are risky (external side effects, irreversible, spends money, sends messages)? A tool you call yourself gets guard_tool. A tool the model picks gets ToolPolicy on ArcjetMiddleware.
2. What limits? (e.g. "10 lookups/min per user" → TokenBucket.)
3. Who is the user for metadata — an opaque user/tenant ID (never PII)? Default: none. The conversation / session id you already have is the correlation id, not the user. Put it on config["configurable"]["arcjet_correlation_id"] or with arcjet_sequence(correlation_id=...).
4. Is an Arcjet outage unacceptable? Every helper defaults to on_guard_error="deny". Ask explicitly about inbound screening before ainvoke: failing closed there means the agent does not run for the duration of the outage, so "allow" is a routine and legitimate choice at that one call site.

The things readers get wrong

1. This is not LangChain JS and not LangGraph JS. Do not import @arcjet/guard/langchain/v1 or @arcjet/guard/langgraph/v1.
2. There is no inbound helper. Screen with core guard() before ainvoke.
3. ArcjetCaptureHandler cannot deny. Policy is guard_tool or ArcjetMiddleware.
4. Correlation is caller-owned, never minted. Do not mint a new id per turn. LangChain's run_id is not used. The config key wins over an enclosing arcjet_sequence; configurable is checked before metadata.
5. Pass tools= the same sequence you gave create_agent. A typo in a policy key is refused at construction instead of leaving that tool unguarded.
6. Key rate limits on the authenticated caller, not a model-supplied order id.
7. success on helper capture is not "the action ran". guard_action / guard_tool / ArcjetMiddleware write metadata.outcome. See the shared Python Guard reference.
8. rules= is a static sequence, not a lambda – a callable raises TypeError. Put per-argument checks in the tool body, or use inputs= (this adapter accepts typed inputs; the resolver is (arguments, config)).
9. A missing decision is not a denial. Verify in Console/CLI.

Step 1: Install and find the guard client

These helpers ship in PyPI arcjet 1.0.0+. Install with the project's package manager. Do not hand-edit requirements.txt with a guessed version.

# guard_tool + capture handlers
pip install "arcjet[langchain]"
# ArcjetMiddleware + ToolPolicy
pip install "arcjet[langchain-agents]"

If the agent has no guard client yet, launch one once at module scope. Use launch_arcjet in async code and launch_arcjet_sync in sync code.

import os
from arcjet.guard import launch_arcjet

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

Step 2: Gate a BaseTool you call — guard_tool

from arcjet.guard.langchain import guard_tool

send_email.args_schema = PublicEmailArgs  # narrow first, then wrap
guarded = guard_tool(
    guard=aj,
    tool=send_email,
    action="email.sent",
    rules=[email_limit(key=user_id, requested=1)],
    on_guard_error="deny",
)

Needs arcjet[langchain]. The result is still a BaseTool. DENY raises ArcjetToolDeniedError; unavailable raises ArcjetToolUnavailableError. If you can name the tool at wiring time, guard_tool is the smaller change.

Step 3: Gate create_agent — ArcjetMiddleware

from langchain.agents import create_agent
from arcjet.guard.langchain import ArcjetMiddleware, ToolPolicy

tools = [send_email, search_orders]
agent = create_agent(
    model="openai:gpt-4o",
    tools=tools,
    middleware=[
        ArcjetMiddleware(
            guard=aj,
            policies={
                "send_email": ToolPolicy(
                    action="email.sent",
                    rules=[email_limit(key=user_id, requested=1)],
                ),
            },
            tools=tools,
            on_guard_error="deny",
        )
    ],
)

await agent.ainvoke(
    {"messages": [...]},
    config={"configurable": {"arcjet_correlation_id": session.id}},
)

Needs arcjet[langchain-agents]. guard= is optional if you already register_arcjet(). They compose: a guarded tool inside a guarded agent evaluates each policy once and both land on the same Sequence.

Step 4: Observe a chain — ArcjetCaptureHandler

from arcjet.guard.langchain import ArcjetAsyncCaptureHandler, ArcjetCaptureHandler

# invoke → ArcjetCaptureHandler; ainvoke → ArcjetAsyncCaptureHandler
chain.invoke(inputs, config={"callbacks": [ArcjetCaptureHandler(guard=aj)]})
await chain.ainvoke(
    inputs, config={"callbacks": [ArcjetAsyncCaptureHandler(guard=aj)]}
)

Same extra as guard_tool. Pair the handler with the call. Neither can deny a call.

Step 5: Screen inbound and correlate

from arcjet.guard import DetectPromptInjection

inbound = DetectPromptInjection()
decision = await aj.guard(
    label="message.received",
    rules=[inbound(user_text)],
)
if decision.conclusion == "DENY":
    raise RuntimeError("message blocked")
if decision.has_failed_open():
    raise RuntimeError("inbound guard unavailable")

Keep a whole run on one Sequence with with arcjet_sequence(correlation_id=session.id): or config={"configurable": {"arcjet_correlation_id": session.id}}. Do not mint a new id per turn.

Verify the integration

1. python -m py_compile (or the project's type-check) passes.
2. Exercise inbound PI (before ainvoke), a guard_tool deny, a middleware deny on a named policy, a capture-only handler that does not block, a rate limit, and fail-closed (an unreachable guard).
3. Confirm in the Arcjet Console / CLI (guards list) that decisions share the caller-owned correlation id.
4. Manual E2E with a real ARCJET_KEY is still-to-verify until you run it.

Worked example: examples/fastapi-langchain-guard. Do not invent a second example name. Do not add an example in this skills repo.