LangChain / LangGraph

New to the adapters? Start with the wrapping overview and comparison table.

Compatible versions: see the compatibility table for the verified langchain range. deepagents rides this adapter but has its own row and range in that table — check the deepagents row, not the langchain one.

wrap_langchain_agent builds a PolicyEnforcer once and installs it on each tool in place (install_enforcer_on_tool) so the same instances inside the compiled graph become policy-gated. It returns a HexgateLangchainAgent proxy that opens a HexgateContext scope and injects a Langfuse callback into every invoke / ainvoke / stream / astream / astream_events call. The hexgate_context is supplied per call, so a single wrapped agent can serve many users concurrently — role resolution happens at call time from the contextvar.

import asyncio
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

from hexgate.runtime import HexgateContext
from hexgate.adapters.langchain import wrap_langchain_agent

INSTRUCTION = (
    "You are a DevOps assistant for a Kubernetes platform. Help engineers read "
    "service logs, restart services, and scale deployments. Pull the service "
    "name, replica count, and environment (dev/staging/prod) from the request "
    "and act directly — the policy layer gates sensitive actions."
)


@tool
def read_logs(service: str, env: str) -> str:
    """Return recent log lines for `service` in `env` (dev/staging/prod)."""
    return f"(stub) {service}@{env}: 200 OK, 200 OK, WARN upstream slow"


@tool
def restart_service(service: str, env: str) -> str:
    """Restart `service` in `env`."""
    return f"(stub) restarted {service}@{env}"


@tool
def scale_deployment(service: str, replicas: int, env: str) -> str:
    """Scale `service` to `replicas` pods in `env`."""
    return f"(stub) scaled {service}@{env} to {replicas} replicas"


TOOLS = [read_logs, restart_service, scale_deployment]


async def main():
    load_dotenv()

    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    graph = create_react_agent(llm, TOOLS, name="devops_agent", prompt=INSTRUCTION)

    agent = wrap_langchain_agent(
        agent=graph,
        tools=TOOLS,          # same list passed to create_react_agent — wrapped in place
        api_key="sk-...",     # or rely on HEXGATE_API_KEY
    )

    result = await agent.ainvoke(
        {"messages": [{"role": "user",
                       "content": "Scale the checkout service to 5 replicas in staging."}]},
        hexgate_context=HexgateContext(user_id="engineer_1", user_roles=["operator"], session_id="session_1"),
    )
    print(result)


if __name__ == "__main__":
    asyncio.run(main())

Under the hood

wrap_langchain_agent builds a PolicySet for the agent, constructs one PolicyEnforcer(policy_set, agent_name=…), and calls install_enforcer_on_tools(tools, enforcer=…) to mutate each tool’s func and coroutine with enforcer-gated closures. handle_tool_error is forced to True. Installation is idempotent — re-installing rebinds the captured originals to the new enforcer without stacking gates.
Each invocation method on HexgateLangchainAgent takes hexgate_context= and opens an async with hexgate_context: (or hexgate_context.sync_scope() for sync) around the delegated CompiledStateGraph call. The active context is pushed onto a contextvar; the guards read it at tool-call time to resolve the matching role’s policy.
A non-allow Decision is rendered as {"ok": False, "error": decision.as_error_payload()} so the LangChain runtime surfaces the structured dict as the tool result instead of raising.
The wrapper also enters propagate_attributes(...) and merges a Langfuse CallbackHandler into the RunnableConfig.callbacks for the duration of the call. Anything not explicitly proxied falls through via __getattr__.

Because LangChain BYO-graph tools are mutated in place by design, the same tools list you pass to create_react_agent flows through already gated — the wrapper holds the policy.

The policy is resolved by agent name, so pass name="devops_agent" to create_react_agent — it must match the agent you registered on the platform. Omit it and the compiled graph keeps LangGraph’s default name ("LangGraph"), so the wrapper resolves the policy for an agent called LangGraph and fails with a 404 unless one is registered under that name.

Runnable example

examples/devops_langchain.py — wrap_langchain_agent (LangChain / LangGraph) end-to-end.

Get started

Concepts

Framework adapters

Policy

CLI

LangChain / LangGraph

Under the hood

Runnable example

​Under the hood

​Runnable example

Under the hood

Runnable example