LangChain Connector

What it does

langchain-noxy 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 by polling the relay's GetDecisionOutcome via the SDK, then applying {"type": "approve"} or {"type": "reject", "message": …} to each tool call.

Flow

┌────────────────┐    tool calls   ┌───────────────────────┐
│ create_agent() │ ──────────────▶ │ NoxyHumanInTheLoop    │
│ + middleware   │                 │   Middleware          │
└────────┬───────┘                 └────────────┬──────────┘
         │ interrupt()                          │ send_decision
         ▼                                      ▼
   ┌───────────┐                           ┌──────────┐
   │Checkpoint │                           │  Noxy    │ ────▶ Devices
   │   state   │                           │  Relay   │
   └───────────┘                           └────┬─────┘
         ▲                                      │ get_decision_outcome
         │ Command(resume=HITLResponse)         │ (poll w/ backoff)
         └────────────── wait_and_resume() ─────┘

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 carrying the decision_id.
5. User approves or rejects on a device, or the decision TTL expires.
6. You call bridge.wait_and_resume(agent, decision_id); the SDK polls GetDecisionOutcome with exponential backoff until the decision settles.
7. The middleware applies {"type": "approve"} or {"type": "reject"} to each suspended tool call and the agent continues.

Relay delivers outcomes via gRPC polling (GetDecisionOutcome); there is no webhook to host. The SDK applies exponential backoff between polls.

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 langchain-noxy

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

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

Configuration

*Identity is passed to NoxyLangChainBridge(client, identity_id) in code; use the env var when your app reads it from the environment.

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 langchain_noxy 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(),
)

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
# agent.invoke(...) suspends with a GraphInterrupt when transfer_funds is proposed;
# read decision_id from the interrupt value.

# Poll relay until approved / rejected / expired (SDK exponential backoff):
final = bridge.wait_and_resume(agent, "<decision_id>")

Manual polling

If you already poll relay elsewhere (a worker, a cron, a separate service), resume from a single get_decision_outcome response instead of running the built-in loop:

from noxy.decision_outcome import WaitForDecisionOutcomeOptions

resume_handler = bridge.create_resume_handler(agent)
response = client.wait_for_decision_outcome(
    WaitForDecisionOutcomeOptions(decision_id="<decision_id>", identity_id="user@example.com")
)
final = resume_handler.resume_from_poll_response(
    response, decision_id="<decision_id>", identity_id="user@example.com"
)

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 langchain_noxy import NoxyAgentState, NOXY_SENT_DECISION_ID_KEY
# NOXY_SENT_DECISION_ID_KEY == "_noxy_sent_decision_id"

Outcome mapping

The connector maps Noxy relay outcomes to LangChain HITL decisions (via hitl_response_from_outcome):

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

Poll tuning

Pass WaitForDecisionOutcomeOptions to bridge.wait_and_resume (the same fields the Python SDK accepts) to control the polling loop:

from noxy.decision_outcome import WaitForDecisionOutcomeOptions

final = bridge.wait_and_resume(
    agent,
    "<decision_id>",
    wait_options=WaitForDecisionOutcomeOptions(
        decision_id="<decision_id>",
        identity_id="user@example.com",
        max_wait_ms=300_000,
    ),
)

API reference

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

NoxyAgentResumeHandler

Helpers and types

FastAPI poll-resume server

The package ships with an end-to-end FastAPI example (examples/poll_resume_server.py): start a run with POST /runs, then resume it with POST /runs/wait. Here is the minimum:

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.decision_outcome import WaitForDecisionOutcomeOptions

from langchain_noxy import NoxyLangChainBridge

IDENTITY = os.environ["NOXY_IDENTITY_ID"]


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


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

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("/runs/wait")
async def wait_for_outcome(body: dict) -> dict:
    decision_id = body.get("decision_id")
    if not decision_id:
        raise HTTPException(status_code=400, detail="decision_id is required")
    options = WaitForDecisionOutcomeOptions(decision_id=str(decision_id), identity_id=IDENTITY)
    try:
        final = await resume_handler.wait_and_resume_async(client, options)
    except Exception as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    return {"ok": True, "result": final}

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

export NOXY_APP_TOKEN="…"
export NOXY_IDENTITY_ID="user@example.com"
uvicorn examples.poll_resume_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.
• Run the poll loop off the request path. wait_and_resume can block for the full max_wait_ms. Run it in a background worker or task queue for long approval windows rather than holding an HTTP request open.
• TTL & poll budget. Tune decision_ttl_seconds on NoxyConfig and max_wait_ms in WaitForDecisionOutcomeOptions: 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 resumed interrupts 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