LangChain adapter: guard the sync paths, pass `isinstance`, and give LangGraph a terminal deny instead of an exception

[zijianyuan-gh]

Package

policy-engine

Description

guard_langchain_tool and guard_langchain_runnable (policy-engine/sdk/python/agent_control_specification/_adapters/langchain.py, lines 51-94 and 23-48) guard ainvoke only. Wiring the guarded tool into an existing LangGraph agent, I hit three things that each need a workaround.

1. Sync paths are blocked rather than guarded. invoke, batch and stream go into the proxy's blocked map (langchain.py:47 and :89-93, built by _generic.py:230-235) and calling any of them raises AdapterUnsupportedError (_shared.py:37-44). That exception is an AgentControlBlocked with reason host_error:adapter_unsupported pinned to intervention point input (_errors.py:6-13), so a plain tool.invoke(...) from sync code reports "blocked input", which sends you looking at the wrong thing. LangGraph's ToolNode calls tool.invoke on its sync path (langgraph/prebuilt/tool_node.py:958), and Runnable composition (RunnableSequence, RunnableParallel) calls invoke, batch and stream on its steps, so anything not fully async is out.

2. The proxy fails isinstance. _ObjectProxy (_shared.py:21-48) forwards attribute access through __getattr__ but is its own class, so isinstance(guarded, BaseTool) and isinstance(guarded_runnable, Runnable) are both False. ToolNode.__init__ (tool_node.py:779-784) checks isinstance(tool, BaseTool) and otherwise passes the object to langchain_core.tools.tool, which raises ValueError: The first argument must be a string or a callable with a __name__ .... Anything else that branches on isinstance(x, Runnable) to decide how to compose behaves the same way.

3. A deny is an exception, and LangGraph treats exceptions as retryable. enforce raises AgentControlBlocked on a deny (_orchestration.py:346-354, :377). With handle_tool_errors=True (the default through langgraph-prebuilt 0.6, tool_node.py at tag 0.6.0 lines 324-326, and still the documented way to keep the loop alive) ToolNode converts it into ToolMessage(status="error", content="Error: AgentControlBlocked(...)\n Please fix your mistakes.") (tool_node.py:111, :1002-1012). The model reads that as "try again", re-issues the same call, the policy denies again, and the loop runs to the recursion limit. With the 1.x default handler (tool_node.py:383-391, which only swallows ToolInvocationError) the exception propagates and the whole graph run aborts. That is at least terminal, but neither path gives the graph a "this tool was denied, carry on without it" outcome.

Proposal:

• Guard the sync paths. Wrap invoke with the same pre/post evaluation through run_sync (already in _host.py:45-77), and derive batch from it. stream can stay blocked if buffering for post-evaluation is out of scope, but the blocked error should say so rather than reporting blocked input.
• Make the proxy pass isinstance. Either build a dynamic subclass of the target's class, or expose __class__ as a property that returns the target's class, which is what wrapt.ObjectProxy and unittest.mock do. The __class__ route is a few lines in _ObjectProxy and keeps it generic for the other adapters.
• Return a structured tool error on deny, or make it configurable: on_deny="raise" | "tool_error". In tool_error mode the guarded tool returns a ToolMessage(status="error") (or a dict with error, message, terminal: true) carrying the verdict reason and message, so the model sees "this tool is not permitted" rather than "fix your mistakes" and a graph can route on it. raise stays the default for backwards compatibility.

Happy to send PRs for the first two. The third probably wants a decision on the default and the payload shape first.

How does this impact your work?

Workaround available, but it is three layers of wrapping. To drop the guarded tool into an existing LangGraph agent I had to re-wrap the proxy in a fresh @tool so ToolNode would accept it, make every caller async, and write a custom handle_tool_errors callable that detects AgentControlBlocked and emits a terminal message instead of the retry prompt.

Timeline

No hard deadline.

Steps to Reproduce

Requires langchain-core and langgraph alongside the Python SDK built from main. Save as repro.py:

import asyncio
from langchain_core.messages import AIMessage
from langchain_core.runnables import Runnable, RunnableLambda
from langchain_core.tools import BaseTool, tool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode
from agent_control_specification import AgentControl, AgentControlBlocked, guard_langchain_runnable, guard_langchain_tool

MANIFEST = """agent_control_specification_version: 0.4.0-alpha.1
metadata:
  name: langchain-repro
policies:
  deny_lookup:
    type: custom
    adapter: repro
intervention_points:
  pre_tool_call:
    policy_target_kind: tool_args
    policy:
      id: deny_lookup
    policy_target: $snap.tool_call.args
"""

class DenyLookup:
    def evaluate(self, invocation):
        return {"decision": "deny", "reason": "tool_not_permitted",
                "message": "lookup is not permitted for this agent."}

@tool
def lookup(q: str) -> str:
    """Look something up."""
    return f"result for {q}"

control = AgentControl.from_native(MANIFEST, None, DenyLookup())
guarded = guard_langchain_tool(control, lookup)

# 1. sync paths
for name, call in [("invoke", lambda: guarded.invoke({"q": "x"})),
                   ("batch", lambda: guarded.batch([{"q": "x"}])),
                   ("stream", lambda: list(guarded.stream({"q": "x"})))]:
    try:
        call(); print(f"{name}: ran")
    except Exception as e:
        print(f"{name}: {type(e).__name__}: {e}  [verdict.message={e.result.verdict.message!r}]")
runnable = guard_langchain_runnable(control, RunnableLambda(lambda x: x))
try:
    runnable.invoke("hi")
except Exception as e:
    print(f"runnable.invoke: {type(e).__name__}: {e}")

# 2. isinstance
print("isinstance(guarded, BaseTool):", isinstance(guarded, BaseTool))
print("isinstance(runnable, Runnable):", isinstance(runnable, Runnable))
try:
    ToolNode([guarded])
except Exception as e:
    print(f"ToolNode([guarded]): {type(e).__name__}: {e}")

# 3. deny inside a graph
@tool
async def lookup_guarded(q: str) -> str:
    """Look something up (re-wrapped so ToolNode accepts it)."""
    return await guarded.ainvoke({"q": q})

call = AIMessage(content="", tool_calls=[{"name": "lookup_guarded", "args": {"q": "x"}, "id": "call-1"}])

def graph_with(node):
    g = StateGraph(MessagesState)
    g.add_node("tools", node)
    g.add_edge(START, "tools")
    g.add_edge("tools", END)
    return g.compile()

out = asyncio.run(graph_with(ToolNode([lookup_guarded], handle_tool_errors=True)).ainvoke({"messages": [call]}))
msg = out["messages"][-1]
print(f"handle_tool_errors=True  -> {type(msg).__name__}(status={msg.status!r}) content={msg.content!r}")
try:
    asyncio.run(graph_with(ToolNode([lookup_guarded])).ainvoke({"messages": [call]}))
except Exception as e:
    print(f"default handler (1.1.0) -> graph run aborts with {type(e).__name__}: {e}")

Expected: invoke/batch evaluate the policy like ainvoke does; the guarded objects pass isinstance; a deny produces a terminal tool error the model does not retry.
Actual: see the log below.

Environment

• Python SDK agent-control-specification 0.4.0b0 built from main at e7f5d2b (2026-09-19); native core as pinned in policy-engine/Cargo.lock (agent-control-spec 0.4.0-alpha.3, agent-hooks-sdk 0.1.0-alpha.5)
• langchain-core 1.6.3, langgraph 1.2.11, langgraph-prebuilt 1.1.0 (tool_node.py line numbers above are from that release; the 0.6.0 default is cited from the 0.6.0 tag)
• Python 3.11.14, macOS 14 (arm64)

Logs / Error Output

invoke: AdapterUnsupportedError: Agent Control Specification blocked input (host_error:adapter_unsupported).  [verdict.message='invoke is not guarded by this adapter; use ainvoke().']
batch: AdapterUnsupportedError: Agent Control Specification blocked input (host_error:adapter_unsupported).  [verdict.message='batch is not guarded by this adapter; use ainvoke().']
stream: AdapterUnsupportedError: Agent Control Specification blocked input (host_error:adapter_unsupported).  [verdict.message='stream is not guarded by this adapter; use ainvoke().']
runnable.invoke: AdapterUnsupportedError: Agent Control Specification blocked input (host_error:adapter_unsupported).
isinstance(guarded, BaseTool): False
isinstance(runnable, Runnable): False
ToolNode([guarded]): ValueError: The first argument must be a string or a callable with a __name__ for tool decorator. Got <class 'agent_control_specification._adapters._shared._ObjectProxy'>
handle_tool_errors=True  -> ToolMessage(status='error') content="Error: AgentControlBlocked('Agent Control Specification blocked pre_tool_call (tool_not_permitted).')\n Please fix your mistakes."
default handler (1.1.0) -> graph run aborts with AgentControlBlocked: Agent Control Specification blocked pre_tool_call (tool_not_permitted).

Code of Conduct

• I agree to follow the Microsoft Open Source Code of Conduct

[github-actions]

Welcome to the Agent Governance Toolkit! Thanks for opening your first issue.
A maintainer will review this shortly. Check our Contributing Guide.
For security issues, use private vulnerability reporting.

[david-long1]

Note: this was opened from a friend's account by mistake while I was filing from their machine. I (david-long1) am the author and will follow up here; happy to send the PRs described. Maintainers, feel free to close and I will refile under my account if you prefer.