LangGraph and CrewAI

LangGraph, CrewAI and the OpenAI Agents SDK are Python frameworks. The Python SDK (pip install theauth) has guards for all three and a generic decorator in theauth.integrations. Any other process that can make an HTTP call can ask theAuth before running a tool with the REST endpoint below. For Node orchestrators such as LangGraph.js there is a TypeScript guard, createHttpGuard.

The endpoint

Mount the theAuth handler (for example at /api/auth) and call it with the agent bearer token.

curl -s -X POST https://YOUR_HOST/api/auth/authorize/token \
  -H "authorization: Bearer $AGENT_TOKEN" \
  -H "content-type: application/json" \
  -d '{"action":"execute","resource":"tool:refund","arguments":{"amount":40}}'

Status	Meaning
200	Allowed. Body `{ "data": { "allowed": true, "auditId": "..." } }`
401	Bad, revoked or expired token. Body `{ "error": { "code", "message" } }` or `data.code` of `invalid_token`, `agent_revoked`, `agent_expired`
403	Denied. Body `{ "data": { "allowed": false, "code", "reason", "auditId" } }`

data.code is the decision code, for example no_permission, rate_limited or approval_required. Treat any non 200 answer, and any network error, as a denial.

Python guards

The Python SDK ships the guards in theauth.integrations. They match the TypeScript helpers: authorize runs before the tool, and a refusal comes back as a structured denial with the decision code and approvalRequired. Nothing is imported from LangGraph, CrewAI or the OpenAI Agents SDK when you import theauth, so the frameworks stay optional.

pip install theauth

Every guard is built on one Gate. Build it once from a client and the agent bearer token. The gate fails closed: a network error, a timeout, a malformed answer or an exception in your own callback becomes a denial with code guard_error, and the tool does not run.

import os

from theauth import TheAuthSyncClient
from theauth.integrations import Gate

client = TheAuthSyncClient(base_url=os.environ["THEAUTH_URL"], timeout=5)  # e.g. https://YOUR_HOST/api/auth
gate = Gate.from_token(client, os.environ["THEAUTH_AGENT_TOKEN"])

Use TheAuthClient for an async app. The same gate then works in async tools, because it checks with await when the tool is a coroutine function. To check by agent id instead of token, pass client.authorize and the id: Gate(client.authorize, "agent_123").

Python (LangGraph)

Wrap the tool function. The signature and docstring are kept, so LangChain still builds the argument schema from them.

from langchain_core.tools import StructuredTool

from theauth.integrations import guard_tool


def refund(order_id: str, amount: float) -> dict:
    return {"order_id": order_id, "refunded": amount}


refund_tool = StructuredTool.from_function(
    func=guard_tool(gate, refund, "refund"),
    name="refund",
    description="Refund an order",
)

A refused call returns the denial dict as the tool message content, so the graph continues and the model can tell the user:

{"ok": false, "denied": true, "toolName": "refund", "code": "approval_required", "reason": "...", "approvalRequired": true, "auditId": "..."}

To guard every tool in one place, wrap the node that executes tool calls. guard_node checks each pending call of the last AI message before the node runs. If one is refused, the node does not run: refused calls get the denial as a tool message and the others get a skipped notice, so every tool_call_id is answered.

from langgraph.prebuilt import ToolNode

from theauth.integrations import guard_node

tool_node = guard_node(ToolNode([refund_tool]), gate)
# graph.add_node("tools", tool_node)

For an async graph use aguard_node with TheAuthClient and Gate.from_token(async_client, token). Prefer guard_tool when a step can hold several calls and you want the allowed ones to run.

Python (CrewAI)

CrewAI tools return text to the model, so give the guard a text formatter. Put it under @tool:

from crewai.tools import tool

from theauth.integrations import guard


@tool("refund")
@guard(gate, "refund", format_denial=lambda d: d.to_text())
def refund_order(order_id: str, amount: float) -> str:
    """Refund an order."""
    return f"Refunded {amount} on {order_id}"

A refusal reads Refused (approval_required): <reason>. Ask the user to approve it. For a tool class that subclasses BaseTool, or a tool you already built, call guard_crewai_tool(tool, gate). It wraps _run or func in place.

Python (OpenAI Agents SDK)

guard_function_tool takes the object @function_tool returns and gives back a copy whose on_invoke_tool checks the gate first. The input JSON is parsed for the argument check, and input that is not valid JSON is refused.

from agents import function_tool

from theauth.integrations import guard_function_tool

@function_tool
def lookup_invoice(invoice_id: str) -> str:
    """Read an invoice."""
    return f"{invoice_id}: paid"

lookup_invoice = guard_function_tool(lookup_invoice, gate)

Pass the result in Agent(tools=[...]) as usual. The denial is returned as JSON text. Use guard_function_tools([...], gate) for a list.

Python (any function)

@guard(gate, "name") works on any sync or async function:

from theauth.integrations import guard

@guard(gate, "send_email")
def send_email(to: str, body: str) -> str:
    return f"sent to {to}"

send_email("a@example.com", "hi")  # returns the denial dict when refused

Parameters named self, cls, ctx, context, config and runtime are treated as run context, not tool input, and are not sent as arguments. Other values are converted to JSON safe form first (Pydantic models and dataclasses become dicts, unknown objects become strings).

Python approvals

When the decision is approval_required, the gate can open a request, wait for a human and retry once with the approved id. Pass request_approval. It receives an ApprovalContext (tool name, agent id, action, resource, arguments, audit id) and returns the approved request id, or None to refuse. wait_for_approval polls client.approvals.get(id) until the request is approved. It returns None when the request is denied, expired or missing, or when timeout_s passes, and the gate then refuses the call. The read route needs an admin user token. The request itself is opened by your backend with theauth.approval.request(...), because POST /approvals on the admin API only opens admin: actions. /api/open-approval below is your own route:

import httpx

from theauth.integrations import wait_for_approval

admin = TheAuthSyncClient(base_url=os.environ["THEAUTH_URL"], token=os.environ["THEAUTH_ADMIN_TOKEN"], timeout=5)


def ask_human(request):
    opened = httpx.post(
        os.environ["APPROVAL_OPENER_URL"],  # your route, e.g. https://YOUR_APP/api/open-approval
        json={
            "agentId": request.agent_id,
            "action": request.action,
            "resource": request.resource,
            "arguments": request.arguments,
        },
        timeout=5,
    )
    return wait_for_approval(admin.approvals, opened.json()["id"], timeout_s=120, poll_interval_s=2)


gate = Gate.from_token(client, os.environ["THEAUTH_AGENT_TOKEN"], request_approval=ask_human)

In an async app use await await_approval(admin_client.approvals, id, timeout_s=120). A sync gate cannot await, so give it a sync request_approval and a sync client. The approved id is spent on the single retry, and the approval is bound to the same action, resource and arguments. See Approval.

TypeScript guard for Node orchestrators

import { createHttpGuard } from "@glinr/theauth/integrations";

const guard = createHttpGuard({
  baseUrl: process.env.THEAUTH_URL!, // e.g. https://YOUR_HOST/api/auth
  token: () => process.env.THEAUTH_AGENT_TOKEN!,
  timeoutMs: 5000,
});

const refund = guard.guard("refund", async (args: { orderId: string; amount: number }) => {
  return { orderId: args.orderId, refunded: args.amount };
});

const out = await refund({ orderId: "ord_1", amount: 40 });
// A refusal resolves to { ok: false, denied: true, toolName, code, reason, approvalRequired }.

Use guard.check(toolName, args) when you want the denial without wrapping a function: it returns null when allowed. The guard never throws. Network errors, timeouts and server errors resolve to a denial with code guard_error. Both authorize routes accept an optional approvalId in the body, which spends an approved request once when the action, resource and arguments match. createHttpGuard does not send one itself. For a retry after approval, call the route yourself with the approved id, or run the check in a Node service that holds theauth.authorize and use the requestApproval option of the Vercel AI SDK helper.

Identity

Authorization

Observability

Controls

Tutorials

LangGraph and CrewAI

The endpoint

Python guards

Python (LangGraph)

Python (CrewAI)

Python (OpenAI Agents SDK)

Python (any function)

Python approvals

TypeScript guard for Node orchestrators

​The endpoint

​Python guards

​Python (LangGraph)

​Python (CrewAI)

​Python (OpenAI Agents SDK)

​Python (any function)

​Python approvals

​TypeScript guard for Node orchestrators

The endpoint

Python guards

Python (LangGraph)

Python (CrewAI)

Python (OpenAI Agents SDK)

Python (any function)

Python approvals

TypeScript guard for Node orchestrators