LangChain

Gate every LangChain tool call through LatchGate’s enforcement pipeline.

Prerequisites

• LatchGate running (latchgate up or production deployment)

Install

pip install latchgate-langchain

Quick start

import asyncio
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from latchgate_langchain import LatchGateToolset

async def main():
    async with await LatchGateToolset.create(
        gate_url="http://localhost:3000",
        agent_id="my-langchain-agent",
    ) as toolset:
        tools = toolset.get_tools()

        llm = ChatOpenAI(model="gpt-4o")
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a helpful assistant with access to gated tools."),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ])
        agent = create_tool_calling_agent(llm, tools, prompt)
        executor = AgentExecutor(agent=agent, tools=tools)

        result = await executor.ainvoke({"input": "Fetch https://httpbin.org/get"})
        print(result["output"])

asyncio.run(main())

How it works

LatchGateToolset.create() discovers all registered LatchGate actions via the REST API and wraps each one as a LangChain BaseTool with the action’s JSON Schema as the tool’s input schema. When the LLM calls a tool, the toolkit routes it through LatchGateClient.execute() — auth, policy, WASM sandbox, verification, and signed receipt all happen transparently.

Filtering actions

toolset = await LatchGateToolset.create(
    gate_url="http://localhost:3000",
    include={"http_fetch", "github_read"},       # only these
    exclude={"send_message"},                     # or skip these
)

UDS transport (production)

from latchgate import LatchGateClient

client = LatchGateClient(socket="/run/latchgate/gate.sock", agent_id="my-langchain-agent")
toolset = await LatchGateToolset.create(client=client)

Audit callback

Use on_audit for programmatic access to receipt metadata after each successful execution:

def my_audit_handler(record):
    # record.action_id, record.receipt_id, record.trace_id, record.verification
    audit_log.store(record)

toolset = await LatchGateToolset.create(
    gate_url="http://localhost:3000",
    on_audit=my_audit_handler,
)

LATCHGATE_URL environment variable

You can set LATCHGATE_URL instead of passing gate_url explicitly:

# With LATCHGATE_URL=http://localhost:3000 set:
toolset = await LatchGateToolset.create()

Discovery timeout

Control the HTTP timeout for the discovery phase (default 15 seconds):

toolset = await LatchGateToolset.create(
    gate_url="http://localhost:3000",
    discovery_timeout=10.0,
)

More details

Full API reference: latchgate-integrations/langchain/README.md.