Skip to content
Agentic Control Plane

Policy & audit for LangChain & LangGraph tools

TL;DR. pip install acp-langchain, add middleware=[ACPMiddleware()] to create_agent(...), bind the end user’s JWT per request — and every LangChain 1.x / LangGraph tool call gets per-user identity, policy enforcement (allow / deny / redact), rate limits, PII detection, and an audit trail. One registration, same control model as Claude Code.

LangChain and LangGraph are the most widely deployed agent frameworks in Python, and LangChain 1.0 made middleware the first-class seam around create_agent — middleware can wrap model calls and tool execution. Out of the box, though, a production deployment still shares one backend API key across every end user’s request — no per-user policy enforcement, no per-user audit trail, no way to tell downstream systems which human triggered which action.

acp-langchain closes that gap on exactly that seam. ACPMiddleware() is one entry in the middleware list; before each tool runs, ACP decides allow / deny / redact based on your workspace’s policy, the end user’s identity, rate limits, and PII detection. Same control model as Claude Code — same /govern/tool-use endpoint, same workspace policies.

Starter · 5-minute install. pip install acp-langchain "langchain>=1.3.3", add middleware=[ACPMiddleware()], bind the JWT per request. See the runnable starter, the control model, or the frameworks index.

Install

A real recording, nothing mocked: the published package installed from PyPI into a clean venv and a live agent run against the gateway. Captured on the 0.1 decorator flow — the install and the run look identical on the 0.2 middleware flow below.
pip install acp-langchain "langchain>=1.3.3"

Minimal agent under policy (langchain.agents.create_agent)

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

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

@tool
def web_search(query: str) -> str:
    """Search the web."""
    return my_search(query)   # your code, your credentials

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

# One registration covers every tool on the agent — present and future.
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[web_search, send_email],
    middleware=[ACPMiddleware()],
)

@app.post("/run")
def run(prompt: str, authorization: str = Header(...)):
    set_context(user_token=authorization.removeprefix("Bearer ").strip())
    result = agent.invoke({"messages": [{"role": "user", "content": prompt}]})
    return {"result": result["messages"][-1].content}

No decorator on either tool. Coverage is a property of the agent, not of each function — there is nothing to forget on the tool you add next month.

create_agent is the LangChain 1.x idiom that replaces the legacy langgraph.prebuilt.create_react_agent; ACPMiddleware requires langchain >= 1.3.3.

What happens on every tool call

  1. ACPMiddleware’s tool-call wrap hook POSTs { tool_name, tool_input, session_id } + Authorization: Bearer <user-jwt> to ACP’s /govern/tool-use.
  2. ACP evaluates workspace policy, user scopes, rate limits, and PII.
  3. Deny → the tool function is never called. The middleware returns a synthetic ToolMessage("tool_error: <reason>"); the model sees the denial as the tool’s output and adapts. The run completes normally.
  4. Allow → your tool runs.
  5. Post-audit: ACP scans the output for PII. redact → the redacted output replaces the original before the model sees it; block → the model sees "[ACP] Blocked: <reason>". Audit row written, rooted in the end user’s identity.

Sync and async agents are both covered (wrap_tool_call / awrap_tool_call).

Composes with LangChain’s own human-in-the-loop

LangChain 1.x ships HumanInTheLoopMiddleware — approve / edit / reject on proposed tool calls, with conditional when predicates since 1.3.3. That’s the pause. ACP is the policy and the ledger: which calls need which treatment, decided from one workspace policy, recorded in one audit trail that also covers Claude Code, Cursor, and CrewAI. They stack in the same list:

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[web_search, send_email],
    middleware=[
        HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),  # the pause
        ACPMiddleware(),                                              # the policy + audit ledger
    ],
)

The HITL interrupt fires after the model proposes calls; ACPMiddleware runs at execution — an approved call still passes the policy check and lands in the audit trail.

Scope coverage when you need to:

ACPMiddleware(tools=["send_email", "delete_record"])   # only these are checked
ACPMiddleware(exclude=["get_time"])                    # everything but these

Legacy stack & custom StateGraph (decorator path)

LangChain 1.0 moved the legacy chains and agents into langchain-classic. On that stack — AgentExecutor, create_tool_calling_agent, the deprecated create_react_agent — or inside a custom StateGraph’s ToolNode, use the v0.1 decorator, which keeps working everywhere:

from langchain_core.tools import tool
from acp_langchain import governed

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

Migrating to 1.x: add middleware=[ACPMiddleware()] to create_agent(...), delete every @governed(...) line, done. Don’t combine both on the same tool, or the call is checked (and audited) twice.

Fail-open

If /govern/tool-use times out (5s default) 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.

Configure your ACP workspace

  1. An IdP configured — ACP verifies the end user’s JWT against your identity provider (Firebase, Auth0, any OIDC). Dashboard → Settings → Identity Provider.
  2. Tools listed — your tool names must match tools enabled in your workspace. Dashboard → Policies → Tools.
  3. Policy — set allow/deny, rate limits, PII mode per tool. Dashboard → Policies.

What shows up in the dashboard

Every checked call appears in cloud.agenticcontrolplane.com/activity with:

  • Actor — the end user’s sub
  • Tool name — the LangChain tool’s .name
  • Decision — allow / deny / redact, with reason
  • Session — groups all tool calls from one request
  • Findings — PII detected in input or output, if any

LangChain/LangGraph tool calls sit alongside Claude Code, Cursor, and CrewAI calls from the same user. One audit log across every agent surface.

Price and meter the model calls

ACPMiddleware is the interception plane — what your agent does. The proxy plane covers what it spends. Both run against the same gateway, and init() wires them together:

import acp_governance as acp

acp.init()            # call before you construct any model client

Model calls now land on the same trail as tool calls, carrying real cost, tokens, and cache economics. Constructing clients explicitly instead? Pass the config rather than calling init():

from acp_governance import model_client_kwargs
from anthropic import Anthropic

client = Anthropic(**model_client_kwargs("anthropic"))

Shapes are "anthropic", "openai" (chat completions), and "openai-responses". The last two are not interchangeable — /v1 serves chat completions, /openai/v1 serves responses.

init() is all-or-nothing per provider: it sets the base URL and the key, or it leaves that provider completely alone and tells you which one it skipped. If OPENAI_BASE_URL already points at your own gateway, ACP won’t silently reroute you.

Your coverage: interception ✓ (middleware) · proxy ✓ (model calls).

Limitations

  • ACPMiddleware needs langchain >= 1.3.3 (the 1.x middleware stack with tool-call wrap hooks). Older or langchain-classic code uses the @governed decorator path above.
  • The two planes are wired separately. The middleware covers tool calls; the proxy covers model calls — each is its own edit, see Price and meter the model calls above. Wiring tools without repointing the model client gives you control with no cost data.
  • Async agents are covered. The middleware implements both sync and async wrap hooks; invoke, ainvoke, and streaming all flow through.
  • Pre-release. acp-langchain@0.2.x. Pin exact versions.

Troubleshooting

Graph runs but nothing appears in the dashboard. Confirm set_context(user_token=...) runs before the agent call — without it the hooks silently no-op — and that ACP has the IdP configured for the JWT’s issuer.

401 from /govern/tool-use. The JWT is invalid, expired, or from an untrusted IdP. Check Settings → Identity Provider.

Tools run but decisions always show allow with reason fail-open. The gateway request is erroring. Check network reachability. Raise timeout via configure(timeout_s=10) if needed.

Policy says deny, but the tool still runs. Confirm ACPMiddleware() is actually in create_agent(middleware=[...]), and that the tool’s .name matches the policy entry. With ACPMiddleware(tools=[...]), names outside the list pass through by design.