LangChain + ACP — Policy & Audit Install Guide

You’ve built a LangChain agent. It works. Now your security team wants to know: who’s using it, what tools are they calling, and what data is flowing through?

acp-langchain gives you a decorator: stack @governed("tool_name") under @tool, and that call gets identity attribution, policy enforcement (allow / deny / redact), PII scanning, rate limiting, and an audit row. Your agent code doesn’t change beyond that one line per tool; your tools’ logic doesn’t change.

What you need

A LangChain agent, create_agent or the legacy stack (Python)
An ACP Cloud workspace (sign up)
For multi-user apps: an identity provider configured in ACP (Auth0, Okta, or Entra ID)

Install

pip install acp-langchain

from fastapi import FastAPI, Header
from langchain.agents import create_agent
from langchain.tools import tool
from acp_langchain import governed, configure, set_context

configure(base_url="https://api.agenticcontrolplane.com")
app = FastAPI()

@tool
@governed("salesforce_query")
def salesforce_query(query: str) -> str:
    """Query Salesforce records using SOQL."""
    return sf.query(query)          # your code, your credentials

@tool
@governed("send_email")
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email on behalf of the user."""
    return sendmail(to, subject, body)

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[salesforce_query, send_email],
)

@app.post("/run")
def run(prompt: str, authorization: str = Header(...)):
    # Bind the end user's JWT per request — every tool call below
    # carries the user's identity to ACP.
    set_context(user_token=authorization.removeprefix("Bearer ").strip())
    result = agent.invoke({"messages": [{"role": "user", "content": prompt}]})
    return {"result": result["messages"][-1].content}

@governed("tool_name") goes inside @tool, closest to the plain function. Coverage is per function — add the decorator to the tools you want checked; a tool without it runs ungoverned.

What happens on every tool call

Pre-check — the @governed wrapper POSTs to ACP /govern/tool-use with the tool name, arguments, and the user JWT bound by set_context.
Decide — ACP evaluates workspace policy, the user’s scopes, rate limits, and PII rules.
Deny → the tool function is never called. The wrapper returns "tool_error: <reason>"; the model sees the denial as the tool’s result and adapts. The run completes normally.
Allow → your tool runs.
Post-audit — the result POSTs to /govern/tool-output. PII scan runs; redact replaces the output before the model sees it, block yields "[ACP] Blocked: <reason>". Audit row written, rooted in the end user’s identity.

Sync and async tools are both covered.

The pause and the brain: composing with LangChain’s own HITL

LangChain 1.x ships HumanInTheLoopMiddleware — interrupt the graph before a sensitive tool runs, and let a human approve, edit, or reject (with conditional when predicates since 1.3.3). That’s the pause, and it’s good: use it.

What it doesn’t give you is the policy and the ledger: which calls should even reach a human, decided consistently from one workspace policy; what was decided, recorded somewhere a security team can query; and the same answers when the same user drives Claude Code, Cursor, or a CrewAI fleet instead. That’s @governed, applied at the function itself — and the two compose:

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[salesforce_query, send_email],   # both already carry @governed(...)
    middleware=[
        HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),  # the pause
    ],
)

The HITL interrupt fires after the model proposes calls; @governed runs at execution — so even a human-approved call still passes the policy check and lands in the audit trail.

On the legacy stack? (`langchain-classic`)

LangChain 1.0 split the package. Legacy chains and agents (AgentExecutor, create_tool_calling_agent, and friends) now live in langchain-classic, and langgraph.prebuilt.create_react_agent is deprecated in favor of create_agent. @governed doesn’t care which constructor calls the decorated function, so the same decorator applies unchanged:

from langchain_core.tools import tool
from acp_langchain import governed

@tool
@governed("salesforce_query")   # policy decorator INSIDE the tool decorator
def salesforce_query(query: str) -> str: ...

It also works inside a custom StateGraph’s ToolNode.

Alternative: consume tools from ACP’s MCP endpoint

If you’d rather ACP hold the backend credentials, connect LangChain to your workspace’s MCP endpoint instead — ACP serves only the tools the authenticated user’s scopes allow, executes them server-side with per-user OAuth tokens, and records every call. See MCP servers for the setup; the decorator path above keeps your tools in your process, the MCP path moves them behind ACP.

Verify in the audit log

Check Activity in the ACP dashboard. Every LangChain tool call shows the actor (the end user’s sub), the tool name, the decision (allow / deny / redact, with reason), the session grouping one request’s calls, and any PII findings. LangChain calls sit alongside Claude Code, Cursor, and CrewAI calls from the same user — one ledger across every agent surface. Export to CSV for compliance reporting.

Fail-open

If /govern/tool-use times out (5s default, configurable) or is unreachable, the tool proceeds with reason "fail-open". Matches Claude Code hook behavior — the policy layer is never a single point of failure for the agent.

Back to guides · Govern a Vercel AI SDK app → · SOC 2 audit trails →