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 = d.payloadHash }
    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 {}
    return d if d.get("effect") == "grant" else None  # pending or reject: do not export

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.