LangChain Connector

What it does

noxy-langchain wraps LangChain's HumanInTheLoopMiddleware and bridges it to the Noxy relay. Drop the middleware on your agent, declare which tools require approval, and the connector handles the rest:

• Intercept guarded tool calls before the agent executes them.
• Send an encrypted actionable to all devices registered for the identity.
• Suspend via LangGraph's interrupt() — the agent state is checkpointed.
• Resume from the Noxy webhook with {"type": "approve"} or {"type": "reject", "message": …} for each tool call.

Flow

┌────────────────┐    tool calls   ┌───────────────────────┐
│ create_agent() │ ──────────────▶ │ NoxyHumanInTheLoop    │
│ + middleware   │                 │   Middleware          │
└────────┬───────┘                 └────────────┬──────────┘
         │ interrupt()                          │ send_decision
         ▼                                      ▼
   ┌───────────┐                           ┌──────────┐
   │Checkpoint │                           │  Noxy    │ ────▶ Devices
   │   state   │                           │  Relay   │
   └───────────┘                           └──────────┘
         ▲                                       │ webhook
         │ Command(resume=HITLResponse)          ▼
         └────────────── resume_from_webhook() ──┘

1. The agent proposes one or more tool calls. The middleware inspects each name against your interrupt_on map.
2. For every guarded call it builds an action_request; one combined Noxy actionable carries all of them.
3. send_decision routes the encrypted actionable; the relay fans it out to all registered devices.
4. interrupt() suspends the agent — control returns to your server with a GraphInterrupt.
5. User approves or rejects on a device, or the decision TTL expires.
6. Noxy posts a webhook to your server. The resume handler maps the outcome to a LangChain HITLResponse and resumes the agent.
7. The middleware applies {"type": "approve"} or {"type": "reject"} to each suspended tool call and the agent continues.

Requirements

• Python 3.10 or newer.
• A LangChain agent built with langchain.agents.create_agent and a checkpointer (required for HITL).
• A Noxy app — see Create App — for the NOXY_APP_TOKEN.
• An identity for the user: phone (E.164), email, your own user id, or wallet address (0x…). See Identity types.

Installation

# Production
pip install noxy-langchain

# With the FastAPI webhook example included
pip install "noxy-langchain[examples]"

Pulls in noxy-sdk >= 2.1.0, langchain >= 0.3, and the matching langgraph runtime as dependencies.

Quick start

import uuid

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from noxy import NoxyConfig, init_noxy_agent_client

from noxy_langchain import NoxyLangChainBridge


@tool
def transfer_funds(to: str, amount: str) -> str:
    """Transfer funds — guarded."""
    return f"Sent {amount} to {to}"


client = init_noxy_agent_client(
    NoxyConfig(
        endpoint="https://relay.noxy.network",
        auth_token="your-app-token",
        decision_ttl_seconds=3600,
    )
)
bridge = NoxyLangChainBridge(client, "user@example.com")  # email, phone, user_id, or 0x…

agent = create_agent(
    init_chat_model("openai:gpt-4o-mini"),
    tools=[transfer_funds],
    middleware=[bridge.create_hitl_middleware({"transfer_funds": True})],
    checkpointer=InMemorySaver(),
)
resume_handler = bridge.create_resume_handler(agent)

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
# agent.invoke(...) suspends with a GraphInterrupt when transfer_funds is proposed.

# Later, in your webhook handler:
final = resume_handler.resume_from_webhook({
    "decisionId": "<decisionId>",
    "identityId": "user@example.com",
    "outcome": "approved",   # or "rejected" / "expired" / "timeout"
})

Configuring the middleware

create_hitl_middleware is a thin wrapper over LangChain's HumanInTheLoopMiddleware. The interrupt_on map decides which tools require approval:

middleware = bridge.create_hitl_middleware(
    {
        # Always require approval
        "transfer_funds": True,

        # Allow Approve, Reject, or Edit args via LangChain's review config
        "deploy_contract": {
            "allowed_decisions": ["approve", "reject", "edit"],
            "description": "Approve contract deployment",
        },

        # Auto-approve in dev, gate in prod
        "rebalance": True if PROD else False,
    },
    description_prefix="Tool execution requires approval",
)

You can mix multiple guarded tools in a single agent step: the middleware bundles them into one decision, and the resume handler returns one {"type": …} entry per call.

Agent state

The middleware extends the default AgentState with one private field — _noxy_sent_decision_id — so re-running the model step after resume does not re-route the decision. You do not need to declare anything yourself; NoxyAgentState is exposed for typing and tests:

from noxy_langchain import NoxyAgentState, NOXY_SENT_DECISION_ID_KEY
# NOXY_SENT_DECISION_ID_KEY == "_noxy_sent_decision_id"

Webhook payload

Noxy posts JSON to your registered webhook:

The connector maps Noxy outcomes to LangChain HITL decisions:

• approved → {"type": "approve"} for every guarded tool call
• rejected → {"type": "reject", "message": "Human rejected the requested agent action."}
• expired / timeout → {"type": "reject", "message": "Decision expired before a human response was received."}

Pass reject_message to hitl_response_from_outcome to override the default copy when you build responses manually.

API reference

NoxyLangChainBridge(client, identity_id, *, registry=None)

NoxyAgentResumeHandler

Helpers and types

FastAPI webhook server

import os
import uuid

from fastapi import FastAPI, HTTPException
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.errors import GraphInterrupt
from noxy import NoxyConfig, init_noxy_agent_client

from noxy_langchain import NoxyLangChainBridge


@tool
def execute_task(task: str) -> str:
    """Execute an approved agent task."""
    return f"Executed: {task}"


client = init_noxy_agent_client(
    NoxyConfig(
        endpoint="https://relay.noxy.network",
        auth_token=os.environ["NOXY_APP_TOKEN"],
        decision_ttl_seconds=3600,
    )
)
bridge = NoxyLangChainBridge(client, os.environ["NOXY_IDENTITY_ID"])

agent = create_agent(
    init_chat_model(os.environ.get("MODEL", "openai:gpt-4o-mini")),
    tools=[execute_task],
    middleware=[bridge.create_hitl_middleware({"execute_task": True})],
    checkpointer=InMemorySaver(),
)
resume_handler = bridge.create_resume_handler(agent)

app = FastAPI()


@app.post("/runs")
def start_run(body: dict) -> dict:
    thread_id = body.get("thread_id") or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}
    try:
        result = agent.invoke(
            {"messages": [{"role": "user", "content": body.get("task", "demo task")}]},
            config,
            version="v2",
        )
        return {"thread_id": thread_id, "result": result, "status": "completed"}
    except GraphInterrupt as exc:
        interrupt_value = exc.args[0][0].value
        return {
            "thread_id": thread_id,
            "status": "awaiting_decision",
            "decision_id": interrupt_value.get("decision_id"),
        }


@app.post("/webhooks/noxy")
async def noxy_webhook(payload: dict) -> dict:
    try:
        final = await resume_handler.resume_from_webhook_async(payload)
    except Exception as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    return {"ok": True, "result": final}

Run it (after pip install "noxy-langchain[examples]"):

export NOXY_APP_TOKEN="…"
export NOXY_IDENTITY_ID="user@example.com"
uvicorn examples.webhook_server:app --reload

Multiple tool calls in one step

A single model turn can produce more than one tool call. The middleware bundles every guarded call into one Noxy decision and stores decision_count in the registry. The resume handler returns one {"type": …} per call so all suspended calls resolve together. The user only sees one approval prompt.

Production checklist

• Persistent checkpointer. Replace InMemorySaver with the LangGraph Postgres or SQLite checkpointer so paused agents survive restarts.
• Persistent registry. The default PendingDecisionRegistry is in-memory. Implement the same three methods (register / lookup / pop) over Redis or your database for multi-process deployments.
• Idempotent webhook handler. Noxy may retry deliveries. resume_from_webhook pops the registry entry, so a retry returns UnknownDecisionError — map it to HTTP 200/410 in your handler.
• TTL. Tune decision_ttl_seconds on NoxyConfig: short for synchronous prompts, long for async approvals. Always handle expired/timeout outcomes (the connector turns them into a reject by default).
• Quota. Each send_decision consumes one decision from your monthly pool — see Pricing. The middleware skips routing when _noxy_sent_decision_id is set, so retried interrupts after resume do not consume more quota.
• Observability. Log decision_id and the LangGraph thread_id at routing time and on resume so you can correlate one approval across the agent, relay, and device.

Troubleshooting

Where to next