LangGraph Connector

What it does

langgraph-noxy is a thin layer over the Noxy Python SDK that turns any LangGraph node into a human-in-the-loop checkpoint. It owns three responsibilities:

• Send — builds an encrypted actionable from your graph state and calls send_decision on the Noxy relay.
• Suspend — calls LangGraph's interrupt() so the graph saves its state to the configured checkpointer and returns control to your server.
• Resume — polls the relay's GetDecisionOutcome via the SDK until the user responds (or the decision expires), then calls graph.invoke(Command(resume=…)) with the decision in state.

Flow

┌───────────┐   send_decision   ┌──────────┐   route   ┌──────────────────────┐
│ LangGraph │ ────────────────▶ │  Noxy    │ ────────▶ │ Devices (web / iOS / │
│   node    │                   │  Relay   │           │ Android / Telegram)  │
└─────┬─────┘                   └────┬─────┘           └─────────┬────────────┘
      │ interrupt()                  │ get_decision_outcome      │ Approve / Reject
      ▼                              │ (poll w/ backoff)         ▼
┌───────────┐                        │                     ┌──────────────┐
│Checkpoint │ ◀──────────────────────┴──────────────────── │  Your server │
│  (state)  │     wait_and_resume() → Command(resume=…)     │   FastAPI…   │
└───────────┘ ─────────────────────────────────────────────▶ └──────────────┘

1. Your graph reaches the HITL node — the node builds an actionable from current state.
2. The connector calls send_decision; the relay encrypts and fans out to every device for the identity.
3. The node calls interrupt(); LangGraph persists the state and returns from graph.invoke with an __interrupt__ marker carrying the decision_id.
4. The user approves or rejects on any device, or the decision TTL expires.
5. You call bridge.wait_and_resume(graph, decision_id) — the SDK polls GetDecisionOutcome with exponential backoff until the decision settles.
6. The handler re-runs the HITL node, which now returns the decision to downstream state via Command(resume=…).

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 LangGraph graph compiled with a checkpointer — required for interrupt() / Command(resume=…).
• A Noxy app — see Create App — for the NOXY_APP_TOKEN.
• An identity for your user: phone (E.164), email, your own user id, or wallet address (0x…). See Identity types.

Installation

# Production
pip install langgraph-noxy

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

The package ships with noxy-sdk >= 2.1.0, langgraph >= 0.2.0, and langgraph-checkpoint >= 2.0.0 as dependencies.

Configuration

Set credentials in your environment or pass them to NoxyConfig:

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

Quick start

End-to-end: a one-node graph that asks the user to approve a task before continuing.

import uuid
from typing import Optional, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import END, START
from langgraph.graph import StateGraph
from noxy import NoxyConfig, init_noxy_agent_client

from langgraph_noxy import NoxyLangGraphBridge, build_tool_call_actionable


class State(TypedDict, total=False):
    task: str
    noxy_decision: Optional[dict]
    _noxy_sent_decision_id: Optional[str]


def build_actionable(state: State) -> dict:
    return build_tool_call_actionable(
        tool="run_task",
        args={"task": state["task"]},
        title="Approve task?",
        summary=state["task"],
    )


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

builder = StateGraph(State)
builder.add_node("noxy_hitl", bridge.create_hitl_node(build_actionable))
builder.add_edge(START, "noxy_hitl")
builder.add_edge("noxy_hitl", END)

graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
paused = graph.invoke({"task": "Send 1 wei"}, config)
decision_id = paused["__interrupt__"][0].value["decision_id"]

# Poll relay until approved / rejected / expired (SDK exponential backoff)
final = bridge.wait_and_resume(graph, decision_id)
print(final["noxy_decision"])  # → {"outcome": "approved", "approved": True, ...}

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(graph)
response = client.wait_for_decision_outcome(
    WaitForDecisionOutcomeOptions(decision_id=decision_id, identity_id=identity)
)
final = resume_handler.resume_from_poll_response(
    response, decision_id=decision_id, identity_id=identity
)

Graph state schema

Two state keys are part of the contract between your graph and the connector:

Always declare both as optional fields on your TypedDict state:

from langgraph_noxy import NOXY_SENT_DECISION_ID_KEY  # "_noxy_sent_decision_id"

class State(TypedDict, total=False):
    task: str
    noxy_decision: Optional[dict]
    _noxy_sent_decision_id: Optional[str]

Building actionables

The HITL node accepts a build_actionable(state) -> dict callable. Use the helper to build a standard propose_tool_call payload:

from langgraph_noxy import build_tool_call_actionable

def build_actionable(state: State) -> dict:
    return build_tool_call_actionable(
        tool="transfer_funds",
        args={"to": "0x000…dEaD", "amountWei": "1"},
        title="Approve transfer of 1 wei?",
        summary="Agent wants to send 1 wei to the burn address.",
        # extra fields are merged verbatim into the actionable
        extra={"chain": "ethereum", "ttlHint": "5m"},
    )

The shape matches the Noxy decision payload — see Decisions & Lifecycle for the full schema.

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(
    graph,
    decision_id,
    wait_options=WaitForDecisionOutcomeOptions(
        decision_id=decision_id,
        identity_id=identity,
        max_wait_ms=300_000,
    ),
)

Timeout / expired outcomes

When the poll budget (max_wait_ms) is exceeded, or the relay reports an expired / timeout outcome, the decision is propagated as a regular decision and your downstream nodes can branch on noxy_decision.outcome. To short-circuit the graph or apply a default action, pass on_timeout to the node:

def on_timeout(state, resume) -> dict:
    # Apply a safe default — e.g. mark the action as skipped
    return {
        "noxy_decision": resume.to_state(),
        "status": "timed_out_default",
    }

bridge.create_hitl_node(build_actionable, on_timeout=on_timeout)

API reference

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

One bridge per identity wires the relay client, the in-memory pending-interrupt registry, the HITL node factory, and the resume handler.

NoxyGraphResumeHandler

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 typing import Optional, TypedDict

from fastapi import FastAPI, HTTPException
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import END, START
from langgraph.graph import StateGraph
from noxy import NoxyConfig, init_noxy_agent_client
from noxy.decision_outcome import WaitForDecisionOutcomeOptions

from langgraph_noxy import NoxyLangGraphBridge, build_tool_call_actionable

IDENTITY = os.environ["NOXY_IDENTITY_ID"]


class State(TypedDict, total=False):
    task: str
    noxy_decision: Optional[dict]
    status: Optional[str]
    _noxy_sent_decision_id: Optional[str]


def build_actionable(state: State) -> dict:
    return build_tool_call_actionable(
        tool="execute_task",
        args={"task": state["task"]},
        title="Approve agent task?",
        summary=f"The agent wants to run: {state['task']}",
    )


def on_timeout(_state, resume) -> dict:
    return {"noxy_decision": resume.to_state(), "status": "timed_out_default"}


def after_decision(state: State) -> dict:
    decision = state.get("noxy_decision") or {}
    if decision.get("approved"):
        return {"status": "executed"}
    if state.get("status") == "timed_out_default":
        return {"status": "skipped_after_timeout"}
    return {"status": "rejected"}


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 = NoxyLangGraphBridge(client, IDENTITY)

builder = StateGraph(State)
builder.add_node("noxy_hitl", bridge.create_hitl_node(build_actionable, on_timeout=on_timeout))
builder.add_node("after_decision", after_decision)
builder.add_edge(START, "noxy_hitl")
builder.add_edge("noxy_hitl", "after_decision")
builder.add_edge("after_decision", END)

graph = builder.compile(checkpointer=InMemorySaver())
resume_handler = bridge.create_resume_handler(graph)

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}}
    state = graph.invoke({"task": body.get("task", "demo task")}, config)
    decision_id = None
    if isinstance(state, dict) and state.get("__interrupt__"):
        decision_id = state["__interrupt__"][0].value.get("decision_id")
    return {"thread_id": thread_id, "decision_id": decision_id, "state": state}


@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, "state": final}

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

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

Production checklist

• Persistent checkpointer. Replace InMemorySaver with the LangGraph Postgres or SQLite checkpointer so paused threads survive restarts.
• Persistent registry. The default PendingInterruptRegistry is in-memory. For multi-process or multi-replica deployments, swap in your own implementation backed by Redis or your database — the contract is the same three methods (register / lookup / pop).
• 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. Set decision_ttl_seconds on NoxyConfig and max_wait_ms in WaitForDecisionOutcomeOptions to sensible bounds — short for synchronous user prompts, long for asynchronous approval workflows. Always handle expired/timeout with on_timeout or a downstream branch.
• Quota. Each send_decision consumes one decision from your monthly pool — see Pricing. The _noxy_sent_decision_id guard prevents re-routing on resume, so polling does not consume extra quota.
• Observability. Log decision_id + thread_id at routing time and on resume so you can correlate a single approval across the agent, relay, and device.

Troubleshooting

Where to next