LangChain / LangGraph

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__.

​

Agent-level enforcement

​

deepagents skills

from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend

graph = create_deep_agent(
    model=llm,
    tools=TOOLS,
    backend=FilesystemBackend(root_dir="./agent_fs", virtual_mode=True),
    skills=["/skills/"],
)
agent = wrap_langchain_agent(agent=graph, tools=TOOLS)

​

How a skill is recognised

​

Limitations

• skill.script: is not supported here. A deepagents script runs through execute, and matching an execute call to a skill’s scripts would mean parsing shell strings. execute is gated on its own terms under tools: instead, on the top-level agent only (a sub-agent’s execute is not gated, see below); a via: [script] rule never fires on this adapter.
• Skills whose paths are not known at wrap time are not gated — a source added mid-run, or a backend populated per invocation. Their reads decide as ordinary read_file calls.
• Sub-agents are not gated. Hexgate gates only the tool objects it wraps: those passed to wrap_langchain_agent(tools=) and those bound on the top-level graph. deepagents’ task tool runs each sub-agent as a separate graph inside the tool call, and any tool a sub-agent runs that is not one of those objects runs with no decision. That includes:
  • the filesystem tools every sub-agent builds on the same backend (ls, read_file, write_file, edit_file, glob, grep, and execute when the backend can run commands), so execute: deny or a denied skill’s SKILL.md holds for the top-level agent only;
  • tools a sub-agent spec declares in its own tools list, unless the same objects are also passed to wrap_langchain_agent(tools=);
  • everything inside a CompiledSubAgent.
  Only the task call itself is recorded. task is a top-level tool gated under tools:, so if a denial must hold inside sub-agents, deny task too, or leave it unlisted under a deny-default policy. That withdraws every sub-agent. Gating sub-agent tools is tracked in #273.
• Not containment. A model can reach a skill’s text another way — grep, ls plus a read of a variant spelling, a copy elsewhere on the backend. This gates the documented activation path on the top-level agent (not inside sub-agents, see above), which is what the audit trail and the kill-switch need. The tool layer remains the boundary.
• The pin is a separate read. The hash is taken just before read_file reads the file itself, so on a writable backend a body changed between the two reads can slip past a pin.

​

Runnable example