Connect the systems you already run

-- kong/plugins/acr/schema.lua
local typedefs = require "kong.db.schema.typedefs"

return {
  name = "acr",
  fields = {
    { protocols = typedefs.protocols_http },
    { config = { type = "record", fields = {
      -- https://<workspace>.agent-cr.com/api/public/v1/gateway
      { gateway_url = { type = "string", required = true } },
      -- A Kong vault reference such as {vault://env/acr-api-key}, never the key itself
      { api_key = { type = "string", required = true, referenceable = true, encrypted = true } },
      { agent_header = { type = "string", default = "X-Agent-Id" } },
      { action_header = { type = "string", default = "X-Acr-Action" } },
      -- "POST /payments/transfers" = "payments.transfer.create"
      { action_map = { type = "map", keys = { type = "string" }, values = { type = "string" }, default = {} } },
      { timeout_ms = { type = "integer", default = 3000 } },
    } } },
  },
}

-- handler.lua (part 1): which action is this request?
local function resolve_action(conf)
  local from_header = kong.request.get_header(conf.action_header)
  if from_header and from_header ~= "" then
    return from_header
  end
  local key = kong.request.get_method() .. " " .. kong.request.get_path()
  return conf.action_map[key]  -- nil when nothing matches
end

-- handler.lua (part 2): one-way fingerprint of the body
local resty_sha256 = require "resty.sha256"
local str = require "resty.string"

local function payload_hash()
  local body = kong.request.get_raw_body() or ""
  local h = resty_sha256:new()
  h:update(body)
  return str.to_hex(h:final())  -- only this hash is sent, never the body
end

-- handler.lua (part 3): ask before Kong forwards
local http = require "resty.http"
local cjson = require "cjson.safe"

local AcrHandler = { PRIORITY = 900, VERSION = "0.1.0" }

local function acr_call(conf, operation, parameters, idem_key)
  local client = http.new()
  client:set_timeout(conf.timeout_ms)
  local res, err = client:request_uri(conf.gateway_url, {
    method = "POST",
    body = cjson.encode({ operation = operation, parameters = parameters }),
    headers = {
      ["content-type"] = "application/json",
      ["x-api-key"] = conf.api_key,  -- resolved from the vault by Kong
      ["idempotency-key"] = idem_key,
    },
  })
  if not res then return nil, err end
  return cjson.decode(res.body)
end

function AcrHandler:access(conf)
  local action = resolve_action(conf)
  if not action then
    return kong.response.exit(403, { message = "No governed action mapped for this route." })
  end
  local hash = payload_hash()
  local params = {
    agent = kong.request.get_header(conf.agent_header),
    action = action,
    resourceRef = kong.request.get_path(),
    payloadHash = hash,
  }
  local idem = "kong-" .. ngx.md5(cjson.encode(params))
  local answer, err = acr_call(conf, "decideAction", params, idem)
  if not answer then
    kong.log.err("ACR unreachable: ", err)
    return kong.response.exit(503, { message = "Not forwarded: no decision." })  -- fail closed
  end
  local d = answer.decision or {}
  if d.effect == "grant" then
    kong.ctx.plugin.decision = { decisionId = d.decisionId, payloadHash = hash }  -- the hash Kong computed
    kong.service.request.set_header("X-Acr-Decision-Id", d.decisionId)
    return  -- forward to the upstream service
  end
  if d.effect == "pending" then
    -- Held for a person. Return once; the caller polls getApprovalStatus. No loop here.
    return kong.response.exit(202, { message = "Held for review.", approvalId = d.approvalId or answer.approvalId })
  end
  return kong.response.exit(403, { message = "Refused.", rationale = d.rationale })
end

-- handler.lua (part 4): report what happened, without delaying the response
local function report(premature, conf, decision, outcome)
  if premature then return end
  local _, err = acr_call(conf, "reportExecutionOutcome", {
    decisionId = decision.decisionId,
    payloadHash = decision.payloadHash,
    outcome = outcome,
  }, "outcome-" .. decision.decisionId)
  if err then kong.log.warn("reportExecutionOutcome failed: ", err) end  -- evidence only
end

function AcrHandler:log(conf)
  local decision = kong.ctx.plugin.decision
  if not decision then return end
  local status = kong.response.get_status()
  local outcome = (status < 500) and "executed" or "downstream_error"
  ngx.timer.at(0, report, conf, decision, outcome)
end

return AcrHandler

# kong.yml (declarative) — enable on the route that agents call
_format_version: "3.0"
services:
  - name: payments
    url: http://payments.internal:8080
    routes:
      - name: transfers
        paths: ["/payments/transfers"]
        plugins:
          - name: acr
            config:
              gateway_url: https://<workspace>.agent-cr.com/api/public/v1/gateway
              api_key: "{vault://env/acr-api-key}"
              action_map:
                "POST /payments/transfers": payments.transfer.create

// acr.ts — server-side only. Never import this into browser code.
const ACR_GATEWAY_URL = process.env.ACR_GATEWAY_URL!; // https://<workspace>.agent-cr.com/api/public/v1/gateway
const ACR_API_KEY = process.env.ACR_API_KEY!;         // kept in your secret store
const ACR_AGENT = process.env.ACR_AGENT!;             // the registered agent this server acts for

type AcrResponse = Record<string, any> & { statusCode: number };

async function acrCall(
  operation: string,
  parameters: Record<string, unknown>,
  idempotencyKey?: string,
): Promise<AcrResponse> {
  const res = await fetch(ACR_GATEWAY_URL, {
    method: "POST",
    headers: {
      "content-type": "application/json",
      "x-api-key": ACR_API_KEY,
      ...(idempotencyKey ? { "idempotency-key": idempotencyKey } : {}),
    },
    body: JSON.stringify({ operation, parameters }),
    signal: AbortSignal.timeout(5000),
  });
  const body = (await res.json()) as AcrResponse;
  if (!res.ok) throw new Error(`ACR ${operation} failed: ${body.error?.code ?? res.status}`);
  return body;
}

type ToolResult = { content: { type: "text"; text: string }[]; isError?: boolean };
type Handler<A> = (args: A) => Promise<ToolResult>;

const blocked = (text: string): ToolResult => ({ content: [{ type: "text", text }], isError: true });

export function governTool<A extends Record<string, unknown>>(
  toolName: string,
  handler: Handler<A>,
  options: { context?: Record<string, unknown> } = {},
): Handler<A> {
  return async (args) => {
    let answer: AcrResponse;
    try {
      answer = await acrCall("decideToolCall", {
        agent: ACR_AGENT,
        toolName,
        toolArguments: args,
        ...(options.context ? { context: options.context } : {}),
      }, crypto.randomUUID());
    } catch {
      // Fail closed: no answer means no execution.
      return blocked("Agent Control Room could not be reached, so the tool did not run.");
    }

    const d = answer.decision;
    if (answer.mcp?.holdToken) {
      // Held for a person. Return once; do not retry in a loop.
      return blocked(`Held for human review. holdToken: ${answer.mcp.holdToken}`);
    }
    if (!answer.mcp?.allowExecution || d?.effect !== "grant") {
      return blocked(`Refused by policy: ${d?.rationale ?? "not allowed"}`);
    }

    let result: ToolResult;
    try {
      result = await handler(args);
    } catch (err) {
      void reportOutcome(d, "downstream_error", "tool_handler_error");
      throw err;
    }
    void reportOutcome(d, result.isError ? "downstream_error" : "executed");
    return result;
  };
}

async function reportOutcome(
  decision: { decisionId: string; payloadHash: string },
  outcome: "executed" | "aborted" | "downstream_error",
  detailCode?: string,
) {
  try {
    await acrCall("reportExecutionOutcome", {
      decisionId: decision.decisionId,
      payloadHash: decision.payloadHash,
      outcome,
      ...(detailCode ? { detailCode } : {}),
    }, `outcome-${decision.decisionId}`);
  } catch (err) {
    // Evidence only: a failed report never changes the tool's result.
    console.error("reportExecutionOutcome failed", err);
  }
}

// Later — for example when the agent or a person asks again — check the hold.
export async function approvalStatus(holdToken: string) {
  const answer = await acrCall("getApprovalStatus", { approvalId: holdToken });
  return answer.approval?.status as
    "pending" | "granted" | "refused" | "expired" | "consumed";
}
// When it is "granted", repeat the identical tool call once:
// that decideToolCall claims the approval and returns the signed grant.

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import { governTool } from "./acr";

const server = new McpServer({ name: "payments", version: "1.0.0" });

server.tool(
  "transfer_funds",
  { fromAccount: z.string(), toAccount: z.string(), amount: z.number() },
  governTool("transfer_funds", async ({ fromAccount, toAccount, amount }) => {
    await bank.transfer(fromAccount, toAccount, amount); // your existing code
    return { content: [{ type: "text", text: "Transfer completed." }] };
  }),
);

# acr.py — server-side only. Requires: pip install langgraph httpx
import os, uuid
import httpx

ACR_GATEWAY_URL = os.environ["ACR_GATEWAY_URL"]  # https://<workspace>.agent-cr.com/api/public/v1/gateway
ACR_API_KEY = os.environ["ACR_API_KEY"]          # kept in your secret store
ACR_AGENT = os.environ["ACR_AGENT"]              # the registered agent this graph acts for

_client = httpx.Client(timeout=5.0)

def acr_call(operation: str, parameters: dict, idempotency_key: str | None = None) -> dict:
    headers = {"content-type": "application/json", "x-api-key": ACR_API_KEY}
    if idempotency_key:
        headers["idempotency-key"] = idempotency_key
    res = _client.post(ACR_GATEWAY_URL, headers=headers,
                       json={"operation": operation, "parameters": parameters})
    body = res.json()
    if res.is_error:
        raise RuntimeError(f"ACR {operation} failed: {body.get('error', {}).get('code', res.status_code)}")
    return body

import functools
from langchain_core.tools import ToolException

def govern_tool(tool_name: str, context_fn=None):
    """Decorator: ask Agent Control Room before the tool body runs."""
    def wrap(fn):
        @functools.wraps(fn)
        def governed(**args):
            params = {"agent": ACR_AGENT, "toolName": tool_name, "toolArguments": args}
            if context_fn:
                params["context"] = context_fn()
            try:
                answer = acr_call("decideToolCall", params, str(uuid.uuid4()))
            except Exception:
                # Fail closed: no answer means no execution.
                raise ToolException("Agent Control Room could not be reached, so the tool did not run.")

            d = answer.get("decision") or {}
            mcp = answer.get("mcp") or {}
            if mcp.get("holdToken"):
                # Held for a person. Return once; do not retry in a loop.
                raise ToolException(f"Held for human review. holdToken: {mcp['holdToken']}")
            if not mcp.get("allowExecution") or d.get("effect") != "grant":
                raise ToolException(f"Refused by policy: {d.get('rationale', 'not allowed')}")

            try:
                result = fn(**args)
            except Exception:
                report_outcome(d, "downstream_error", "tool_handler_error")
                raise
            report_outcome(d, "executed")
            return result
        return governed
    return wrap

def report_outcome(decision: dict, outcome: str, detail_code: str | None = None) -> None:
    params = {"decisionId": decision["decisionId"], "payloadHash": decision["payloadHash"], "outcome": outcome}
    if detail_code:
        params["detailCode"] = detail_code
    try:
        acr_call("reportExecutionOutcome", params, f"outcome-{decision['decisionId']}")
    except Exception as err:
        # Evidence only: a failed report never changes the tool's result.
        print("reportExecutionOutcome failed", err)

from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition

@tool
@govern_tool("transfer_funds")
def transfer_funds(from_account: str, to_account: str, amount: float) -> str:
    """Transfer money between two accounts."""
    bank.transfer(from_account, to_account, amount)  # your existing code
    return "Transfer completed."

tools = [transfer_funds]
graph = StateGraph(MessagesState)
graph.add_node("agent", call_model)  # your existing model node, bound to tools
graph.add_node("tools", ToolNode(tools, handle_tool_errors=True))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", tools_condition)
graph.add_edge("tools", "agent")
app = graph.compile(checkpointer=checkpointer)

from langgraph.types import interrupt

def wait_for_person(hold_token: str) -> str:
    """Pause the graph until someone resumes it, then check the hold once."""
    interrupt({"holdToken": hold_token})  # the graph stops here; resume with Command(resume=...)
    answer = acr_call("getApprovalStatus", {"approvalId": hold_token})
    return answer["approval"]["status"]  # pending | granted | refused | expired | consumed
# When it is "granted", repeat the identical tool call once:
# that decideToolCall claims the approval and returns the signed grant.

# acr_scan.py — server-side only.
# Requires: pip install presidio-analyzer httpx && python -m spacy download en_core_web_lg
import os, json, hashlib, uuid
from datetime import datetime, timezone
import httpx

ACR_GATEWAY_URL = os.environ["ACR_GATEWAY_URL"]  # https://<workspace>.agent-cr.com/api/public/v1/gateway
ACR_API_KEY = os.environ["ACR_API_KEY"]          # kept in your secret store
ACR_AGENT = os.environ["ACR_AGENT"]              # the registered agent this service acts for

_client = httpx.Client(timeout=5.0)

def acr_call(operation: str, parameters: dict, idempotency_key: str | None = None) -> dict:
    headers = {"content-type": "application/json", "x-api-key": ACR_API_KEY}
    if idempotency_key:
        headers["idempotency-key"] = idempotency_key
    res = _client.post(ACR_GATEWAY_URL, headers=headers,
                       json={"operation": operation, "parameters": parameters})
    body = res.json()
    if res.is_error:
        raise RuntimeError(f"ACR {operation} failed: {body.get('error', {}).get('code', res.status_code)}")
    return body

from presidio_analyzer import AnalyzerEngine

_analyzer = AnalyzerEngine()
FINANCIAL = {"CREDIT_CARD", "IBAN_CODE", "US_BANK_NUMBER"}
CONTACT = {"PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER", "LOCATION"}
SPECIAL = {"MEDICAL_LICENSE"}  # extend with your own recognizers

def fact(value, asserted_by: str) -> dict:
    """Each fact names its own source and time."""
    return {"value": value, "assertedBy": asserted_by,
            "observedAt": datetime.now(timezone.utc).isoformat()}

def presidio_facts(text: str) -> dict:
    try:
        found = {r.entity_type for r in _analyzer.analyze(text=text, language="en") if r.score >= 0.6}
    except Exception:
        # A failed scan is unknown, never clean.
        return {"sec.dlp_verdict": fact("not_scanned", "presidio-analyzer")}
    band = ("special_category" if found & SPECIAL else "financial" if found & FINANCIAL
            else "contact" if found & CONTACT else "none")
    return {
        "sec.pii_detected": fact("true" if found else "false", "presidio-analyzer"),
        "sec.pii_categories_band": fact(band, "presidio-analyzer"),
        "sec.dlp_verdict": fact("flagged" if found else "clean", "presidio-analyzer"),
    }

def ask_before_export(text: str, destination: str, context: dict) -> dict | None:
    payload_hash = hashlib.sha256(text.encode("utf-8")).hexdigest()  # content never leaves you
    body = {"agent": ACR_AGENT, "action": "data.export", "resourceRef": destination,
            "payloadHash": payload_hash, "context": context}
    key = hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()
    try:
        answer = acr_call("decideAction", body, key)
    except Exception:
        return None  # Fail closed: no answer means no export.
    d = answer.get("decision") or {}
    if d.get("effect") != "grant":
        return None  # pending or reject: do not export
    return {**d, "payloadHash": payload_hash}  # keep your own hash for the outcome report

def export_with_scan(text: str, destination: str) -> str:
    decision = ask_before_export(text, destination, presidio_facts(text))
    if decision is None:
        return "Not exported: refused, held for a person, or no answer."
    try:
        storage.export(destination, text)  # your existing code
        outcome = "executed"
    except Exception:
        outcome = "downstream_error"
    try:
        acr_call("reportExecutionOutcome", {"decisionId": decision["decisionId"],
                 "payloadHash": decision["payloadHash"], "outcome": outcome},
                 f"outcome-{decision['decisionId']}")
    except Exception as err:
        print("reportExecutionOutcome failed", err)  # evidence only
    return outcome

# Example rules you write in your workspace for data.export (not in code):
# 1. Refuse when sec.pii_categories_band is special_category.
# 2. Refuse when sec.dlp_verdict is not_scanned.
# 3. Ask a person first when sec.dlp_verdict is flagged.
# 4. Otherwise allow.

// identity.ts — server-side only. Requires: npm install jose
// Checking a token needs only your provider's public keys: no client secret here.
const IDP_ISSUER = process.env.IDP_ISSUER!;     // Entra ID: https://login.microsoftonline.com/<tenant-id>/v2.0
                                                // Okta:     https://<your-org>.okta.com/oauth2/default
const IDP_JWKS_URL = process.env.IDP_JWKS_URL!; // Entra ID: https://login.microsoftonline.com/<tenant-id>/discovery/v2.0/keys
                                                // Okta:     https://<your-org>.okta.com/oauth2/default/v1/keys
const IDP_AUDIENCE = process.env.IDP_AUDIENCE!; // your own API's client ID or application ID URI
const IDP_ROLE_CLAIM = process.env.IDP_ROLE_CLAIM ?? "roles"; // Entra app roles: "roles"; Okta: usually "groups"
const FINANCE_ROLE = process.env.FINANCE_ROLE!; // for example Finance.Approver
const ASSERTER = "identity-verifier";           // the asserter name your context contract accepts

import { createRemoteJWKSet, jwtVerify, type JWTPayload } from "jose";

// Fetches and caches the provider's public keys and follows key rotation.
const jwks = createRemoteJWKSet(new URL(IDP_JWKS_URL));

export async function verifyToken(authorization: string | null): Promise<JWTPayload> {
  const token = authorization?.match(/^Bearer (.+)$/i)?.[1];
  if (!token) throw new Error("no_token");
  const { payload } = await jwtVerify(token, jwks, {
    issuer: IDP_ISSUER,
    audience: IDP_AUDIENCE,
    algorithms: ["RS256"],
    requiredClaims: ["iat", "exp"],
    clockTolerance: 60,
  });
  return payload; // signature, issuer, audience and expiry all checked — or it threw
}

type Fact = { value: string | boolean; assertedBy: string; observedAt: string };

export function identityFacts(payload: JWTPayload): Record<string, Fact> {
  // When the provider issued these claims, so a contract can turn away old sign-ins.
  const observedAt = new Date(payload.iat! * 1000).toISOString();
  const fact = (value: string | boolean): Fact => ({ value, assertedBy: ASSERTER, observedAt });
  const roles = payload[IDP_ROLE_CLAIM];
  // Entra ID delegated tokens carry scp; Okta user tokens carry uid. Check your own provider's tokens.
  const isPerson = typeof payload.scp === "string" || typeof payload.uid === "string";
  return {
    "requester.kind": fact(isPerson ? "person" : "service"),
    "requester.finance_role": fact(Array.isArray(roles) && roles.includes(FINANCE_ROLE)),
  };
  // Send what the rules need — not names, email addresses or the token itself.
}

import { createHash, randomUUID } from "node:crypto";
// acrCall and ACR_AGENT: the same configuration and gateway helper as the MCP recipe, steps 1–2.

export async function commitTransaction(req: Request): Promise<Response> {
  let context: Record<string, unknown>;
  try {
    context = identityFacts(await verifyToken(req.headers.get("authorization")));
  } catch {
    // Fail closed: a missing, invalid or expired token means no request and no work.
    return new Response("Sign-in could not be verified.", { status: 401 });
  }
  const body = await req.json();                           // the step the agent proposed
  context["effect.amount_band"] = amountBand(body.amount); // your own thresholds

  let answer;
  try {
    answer = await acrCall("decideAction", {
      agent: ACR_AGENT,
      action: "transaction.commit",
      payloadHash: createHash("sha256").update(JSON.stringify(body)).digest("hex"),
      context,                                             // verified facts only
    }, randomUUID());
  } catch {
    return new Response("No decision, so nothing ran.", { status: 503 }); // fail closed
  }
  const effect = answer.decision?.effect;
  if (effect === "pending") return new Response("Held for a person.", { status: 202 });
  if (effect !== "grant") return new Response("Refused by policy.", { status: 403 });
  // ...run the transaction and report the outcome as in the other recipes
}

# Context contract for transaction.commit, set in your workspace (not in code):
# requester.kind          agreed list: person, service   required, provenance from identity-verifier, at most 5400 s old
# requester.finance_role  yes/no                         required, provenance from identity-verifier, at most 5400 s old
# Mode: enforced — a request without valid facts is undecidable: neither allowed nor refused.
#
# Example rules for transaction.commit:
# 1. Refuse when requester.kind is service.
# 2. Ask a person first when effect.amount_band is high.
# 3. Allow when requester.finance_role is true and effect.amount_band is low or medium.
# 4. Otherwise refuse (the default).