Use with your existing stack

All docs

Add CAIN to the stack you already have#

You do not have to replace anything. Keep your framework, your model gateway and your tracing tool. CAIN adds one thing they do not: a decision before the tool runs, enforced, and a signed record of it afterwards.

You already use	It answers	CAIN adds
An agent framework (LangChain, LangGraph, OpenAI Agents SDK, CrewAI, MCP)	How the agent plans and calls tools	Whether this call may run, right now, for this agent
A tracing or evaluation tool	What happened, after it happened	A refusal before it happens, and a signed record either way
An LLM gateway or prompt guardrail	What goes into and out of the model	What the model is allowed to *do* with your tools

Everything on this page is plain HTTP and the Python standard library. There is nothing to install and no SDK to adopt. Get a key at /signup (free, no card), then export CAIN_API_KEY=....

The guard: one decorator for every framework#

Put this file next to your agent code. It asks CAIN before the function runs and refuses unless the decision permits it: verdict starts with ALLOWED (ALLOWED, ALLOWED_DEGRADED, or ALLOWED_WITH_DENIALS while a workspace is in shadow mode) and blocked is false. REQUIRE_APPROVAL, BLOCKED, a timeout and an unreachable service all mean "do not run".

# cain_guard.py
import functools, json, os, urllib.error, urllib.request

CAIN_URL = os.environ.get("CAIN_URL", "https://cainstudio.online") + "/fabric/decisions"

class CainRefused(Exception):
    """The call was not allowed. The message says why; the tool did not run."""

def allowed(d):
    return d.get("blocked") is False and str(d.get("verdict", "")).startswith("ALLOWED")

def cain_decide(tool, payload, agent_id):
    body = json.dumps({"path": f"/tools/{tool}", "payload": payload, "agent_id": agent_id}).encode()
    req = urllib.request.Request(CAIN_URL, data=body, headers={
        "X-API-Key": os.environ["CAIN_API_KEY"], "content-type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=5) as r:
            return json.load(r)
    except urllib.error.HTTPError as e:         # a refusal can come with a non-2xx status: keep its reason
        try:
            return {"verdict": "ERROR", **json.load(e)}
        except Exception:
            return {"verdict": "ERROR", "detail": f"HTTP {e.code}"}
    except Exception as e:                      # unreachable, timeout, bad JSON: fail closed
        return {"verdict": "ERROR", "detail": str(e)}

def guarded(tool=None, agent_id="my-agent"):
    """Decorator: ask CAIN before the function runs. Keeps the function's name, docstring and
    signature, so framework decorators applied on top still see the original tool."""
    def wrap(fn):
        name = tool or fn.__name__
        @functools.wraps(fn)
        def inner(*args, **kwargs):
            d = cain_decide(name, kwargs or {"args": [repr(a) for a in args]}, agent_id)
            if not allowed(d):
                raise CainRefused(f"{name}: {d.get('verdict')} (decision {d.get('decision_id', '-')})")
            return fn(*args, **kwargs)
        return inner
    return wrap

The pattern is always the same: @guarded(...) goes directly on your function, and the framework's own decorator goes on top of it.

*Tested:* the LangChain, OpenAI Agents SDK and MCP examples below were run against langchain-core, openai-agents and mcp 2.2 with a local stand-in for the decision API (allowed calls run with the same tool name, description and argument schema; refused and unreachable calls do not run). The CrewAI example uses the same mechanism but was not run.

LangChain and LangGraph#

from langchain_core.tools import tool
from cain_guard import guarded

@tool
@guarded("send_email", agent_id="support-bot")
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a customer."""
    return mailer.send(to, subject, body)

The model sees the same name, description and arguments as before. LangGraph's ToolNode calls the same tool object, so a graph gets the check with no other change. When CAIN refuses, CainRefused is raised inside the tool call: return it to the model as the tool result (for example with ToolNode(tools, handle_tool_errors=True)) so the agent learns the call was refused instead of crashing.

OpenAI Agents SDK#

from agents import Agent, function_tool
from cain_guard import guarded

@function_tool
@guarded("refund_order", agent_id="billing-agent")
def refund_order(order_id: str, amount_cents: int) -> str:
    """Refund part or all of an order."""
    return payments.refund(order_id, amount_cents)

agent = Agent(name="Billing", instructions="Help with billing.", tools=[refund_order])

function_tool builds its JSON schema from the original signature, which functools.wraps preserves.

CrewAI#

from crewai.tools import tool
from cain_guard import guarded

@tool("Delete a file")
@guarded("delete_file", agent_id="ops-crew")
def delete_file(path: str) -> str:
    """Delete a file from the shared workspace."""
    os.remove(path)
    return f"deleted {path}"

An MCP server you own (Python)#

from mcp.server.mcpserver import MCPServer      # mcp 2.x; on mcp 1.x: from mcp.server.fastmcp import FastMCP
from cain_guard import guarded

mcp = MCPServer("files")

@mcp.tool()
@guarded("write_file", agent_id="mcp-files")
def write_file(path: str, content: str) -> str:
    """Write a file."""
    open(path, "w").write(content)
    return "ok"

For MCP servers you do not own, put MCPGate in the call path instead: see MCP.

Anything else (TypeScript, Go, a queue worker)#

It is one HTTP call. Run the tool only when verdict starts with ALLOWED and blocked is false:

curl -s https://cainstudio.online/fabric/decisions \
  -H "X-API-Key: $CAIN_API_KEY" -H 'content-type: application/json' \
  -d '{"path":"/tools/send_email","agent_id":"support-bot","payload":{"to":"a@example.com"}}'

Every endpoint, with its request and response schema, is in the OpenAPI spec.

Prefer a package? (coming to PyPI)#

The same guard, plus a LangChain helper and a CLI, is packaged as cainstudio (standard library only). It is not on PyPI yet; until it is, install the hosted wheel:

# pip install https://cainstudio.online/cainstudio-0.2.0-py3-none-any.whl
# LangChain: pip install "cainstudio[langchain] @ https://cainstudio.online/cainstudio-0.2.0-py3-none-any.whl"
# cainstudio try                          # no key: a real live decision, stage by stage
import cainstudio

@cainstudio.guard()        # raises ActionBlocked / ApprovalRequired / CainUnavailable; never runs on error
def send_email(to: str, subject: str): ...

from cainstudio.langchain import protect
tools = protect([search, send_email], agent_id="support-agent")

Then control it without redeploying#

Once your tools ask CAIN, you change behaviour from the console or the API, not in code:

• Tool rules: deny delete_file outright, or require a human approval for refund_order above a limit.
• Approvals: held calls wait for someone on your team to approve or deny them.
• Kill switch: halt every agent in the workspace with one call.
• Traces: each agent run, step by step, with the decision behind every tool call.
• Evidence: every decision is recorded and Ed25519-signed, and you can verify it later.

Moving from a prompt-only guardrail#

If today's protection is a filter on the prompt, keep it. The model can still be talked into calling a tool. The guard above runs at the tool itself, so it applies whatever the prompt said. Try the difference without an account:

curl -s -X POST 'https://cainstudio.online/fabric/try?scenario=prompt-injection'

Evaluating CAIN for a team? [Contact us](mailto:support@cainstudio.online) and we will put it in front of one of your agent's real tool calls with you.