pip install "arcjet[strands-agents] @ git+https://github.com/arcjet/arcjet-py.git@a630806169b92757192f3f5cce2e305827b26567"

import os
from arcjet.guard import launch_arcjet

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

from strands import Agent, tool
from arcjet.guard import DetectPromptInjection, TokenBucket, launch_arcjet
from arcjet.guard.strands_agents import guard_hooks, guard_tool, strands_agent_context

aj = launch_arcjet(key=os.environ["ARCJET_KEY"])
lookup_limit = TokenBucket(
    label="order.looked-up",
    bucket="lookups",
    refill_rate=10,
    interval_seconds=60,
    max_tokens=10,
)
mcp_limit = TokenBucket(
    label="mcp.invoked",
    bucket="mcp-access",
    refill_rate=20,
    interval_seconds=60,
    max_tokens=20,
)
inbound = DetectPromptInjection()
user_id = authenticated_user_id

@tool  # event.interrupt() is HITL — not this policy gate
def lookup_order(order_id: str) -> dict:
    """Look up an order by ID."""
    return {"order_id": order_id, "status": "shipped"}

lookup_order = guard_tool(
    guard=aj,
    tool=lookup_order,
    action="order.looked-up",
    rules=[lookup_limit(key=user_id, requested=1)],
    on_guard_error="deny",
)

mcp_tools = []  # from an MCP client you did not wrap with guard_tool
agent = Agent(
    tools=[lookup_order, *mcp_tools],
    # The agent-wide gate for tools guard_tool did not wrap.
    # Already-wrapped tools are skipped, so Guard is not called twice.
    # BeforeToolCallEvent.cancel_tool is True or a str (JSON of the payload).
    hooks=[
        guard_hooks(
            guard=aj,
            action="mcp.invoked",
            rules=[mcp_limit(key=user_id, requested=1)],
            session_id=conversation_id,
            on_guard_error="deny",
        ),
    ],
)

invocation_state = {"sessionId": conversation_id}
derived = strands_agent_context(invocation_state)
decision = await aj.guard(
    label="message.received",
    rules=[inbound(user_text)],
    correlation_id=derived.correlation_id,
)
if decision.conclusion == "DENY":
    raise RuntimeError("message blocked")
if decision.has_failed_open():
    raise RuntimeError("inbound guard unavailable")

agent(user_text, invocation_state=invocation_state)