Gate for LangChain & LangGraph Developers

LangChain and LangGraph agents write to production systems — CRM, email, databases, APIs. Gate adds a deterministic approval layer to every write in 10 lines of code. No LLM at evaluation time. Full audit trail.

The problem with LangGraph agent writes

LangGraph agents execute graph nodes that trigger real-world side effects. A node sends email. Another writes to Salesforce. A third issues a Stripe refund. LangGraph handles the orchestration — nothing handles the writes.

The LangChain ecosystem provides tracing (LangSmith), prompt management, and memory. It does not provide write-path governance. Writes happen when the node executes. There is no checkpoint between "agent decided to write" and "write landed in production."

Gate is that checkpoint.

How Gate integrates with LangGraph

Gate replaces direct tool calls with an intent-first pattern. The agent submits what it wants to do. Gate evaluates the YAML policy. Only an approved intent triggers execution.

from langchain_core.tools import tool

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email directly."""
    # This fires immediately — no review, no audit
    return gmail_client.send(to=to, subject=subject, body=body)

from langchain_core.tools import tool
from zehrava_gate import Gate

gate = Gate(endpoint="https://your-gate.com", api_key="gate_sk_...")

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email through Gate — evaluated, logged, approved."""
    p = gate.propose(
        payload=f"To: {to}\nSubject: {subject}\n\n{body}",
        destination="gmail.send",
        policy="outbound-email",
        record_count=1
    )
    if p["status"] == "blocked":
        return f"Blocked: {p['blockReason']}"
    if p["status"] == "duplicate_blocked":
        return f"Duplicate — already submitted: {p['blockReason']}"
    if p["status"] == "pending_approval":
        return f"Queued for human review. Intent ID: {p['intentId']}"

    # approved — request execution order
    order = gate.execute(intent_id=p["intentId"])
    return f"Approved. Execution token: {order['execution_token'][:12]}…"

LangGraph StateGraph — full example

Gate integrates with any LangGraph StateGraph node. The pattern replaces direct side-effect calls with Gate-governed writes.

from langgraph.graph import StateGraph, END
from zehrava_gate import Gate, GateError
from typing import TypedDict

gate = Gate(endpoint="https://your-gate.com", api_key="gate_sk_...")

class AgentState(TypedDict):
    lead_id: str
    email_body: str
    gate_intent_id: str
    status: str

def draft_email(state: AgentState) -> AgentState:
    """LLM drafts the email body."""
    body = llm.invoke(f"Write follow-up email for lead {state['lead_id']}")
    return {**state, "email_body": body.content}

def gate_submit(state: AgentState) -> AgentState:
    """Submit intent to Gate — never writes directly."""
    p = gate.propose(
        payload=state["email_body"],
        destination="gmail.send",
        policy="outbound-email",
        record_count=1,
        idempotency_key=f"lead-email-{state['lead_id']}"  # blocks retry loops
    )
    if p["status"] in ["blocked", "duplicate_blocked"]:
        return {**state, "status": "blocked", "gate_intent_id": p["intentId"]}
    return {**state, "status": p["status"], "gate_intent_id": p["intentId"]}

def route_after_gate(state: AgentState) -> str:
    if state["status"] == "approved":
        return "execute"
    return END  # blocked or pending — halt

def execute_write(state: AgentState) -> AgentState:
    """Execute only on Gate approval."""
    order = gate.execute(intent_id=state["gate_intent_id"])
    # Pass order["execution_token"] to your worker — it performs the actual send
    return {**state, "status": "succeeded"}

graph = StateGraph(AgentState)
graph.add_node("draft_email", draft_email)
graph.add_node("gate_submit", gate_submit)
graph.add_node("execute_write", execute_write)
graph.set_entry_point("draft_email")
graph.add_edge("draft_email", "gate_submit")
graph.add_conditional_edges("gate_submit", route_after_gate, {"execute": "execute_write", END: END})
graph.add_edge("execute_write", END)

What the policy looks like

Gate evaluates a YAML policy file. No LLM processes the policy at evaluation time. The same input always produces the same output.

# policies/outbound-email.yaml
id: outbound-email
destinations:
  - gmail.send
  - sendgrid.send
  - loops.send
block_if_terms:
  - "refund guaranteed"
  - "legal action"
  - "sue"
  - "settlement"
auto_approve_under: 1         # single emails auto-approve
require_approval_over: 10     # bulk sends require human review
expiry_minutes: 60

Why Gate instead of LangSmith interrupt?

Install in 60 seconds

# Start Gate server
npx zehrava-gate --port 4000 --policy-dir ./policies

# Register your LangGraph agent
curl -X POST http://localhost:4000/v1/agents/register \
  -H "Content-Type: application/json" \
  -d '{"name":"langgraph-agent","riskTier":"standard","framework":"langgraph"}'

# Install Python SDK
pip install zehrava-gate

The 3 patterns Gate prevents in LangGraph

1. Retry loop duplication

LangGraph retry logic re-executes failed nodes. Idempotency keys block duplicate writes — same agent, same key, Gate returns duplicate_blocked instead of writing twice.

2. Policy drift under context pressure

LLM behavior drifts across long conversations. A 64k-token context produces different decisions than a fresh session. Gate's YAML evaluation is stateless — the 100,000th evaluation is identical to the first.

3. Undetected unauthorized destinations

Agents hallucinate destination strings. A policy with destinations: [gmail.send, sendgrid.send] blocks writes to gmail.send.cc.attacker.com or any string not in the allowlist. Destination validation happens at Gate, not in the LLM prompt.

Gate is MIT-licensed and self-hosted. Run it inside your VPC alongside your LangGraph deployment.