Noxy LangGraph Connector

LangGraph connector for Noxy human-in-the-loop guardrails. Pauses agent graphs with interrupt(), routes encrypted approval prompts to all devices registered for the identity (web, iOS, Android, Telegram), and resumes execution when Noxy fires a webhook.

Flow

sequenceDiagram
    participant G as LangGraph
    participant N as Noxy Relay
    participant D as User Devices
    participant S as Your Server

    G->>N: send_decision
    N->>D: Route to registered devices (web, iOS, Android, Telegram)
    G->>G: interrupt() — state saved to checkpointer
    Note over G: Graph suspended

    alt User responds
        D->>N: Approve / Reject
        N->>S: Webhook (outcome)
        S->>G: Command(resume=decision)
        G->>G: Continue with decision in state
    else Timeout
        N->>S: Webhook (timeout / expired)
        S->>G: Command(resume=timeout)
        G->>G: Continue with default behaviour
    end

1. Graph reaches the HITL node.
2. Noxy routes an encrypted actionable to all devices registered for the identity (web, iOS, Android, Telegram).
3. The node calls interrupt() — LangGraph suspends and persists state via a checkpointer.
4. User responds on any registered device or the decision TTL expires.
5. Noxy fires a webhook to your server.
6. Your server calls NoxyGraphResumeHandler.resume_from_webhook() → Command(resume=...).
7. The graph continues with the human decision (or timeout default) in state.

Requirements

• Python >= 3.10
• A LangGraph graph compiled with a checkpointer (required for interrupt())
• noxy-sdk credentials (NOXY_APP_TOKEN, target identity)

Target identity can be a phone number, email, user id, or wallet address (0x…).

Installation

pip install noxy-langgraph

For the FastAPI webhook example:

pip install "noxy-langgraph[examples]"

Local development against the monorepo SDK:

pip install -e ../../sdks/python-sdk
pip install -e ".[dev,examples]"

Quick start

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 noxy_langgraph 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,
    )
)
# Phone, email, user id, or wallet address
bridge = NoxyLangGraphBridge(client, "user@example.com")

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())
resume_handler = bridge.create_resume_handler(graph)

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

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

Graph state

Include optional _noxy_sent_decision_id in your state schema. The resume handler sets it via Command(update=...) so the HITL node does not re-route the decision when LangGraph re-executes the node after resume (LangGraph always re-runs the node body from the top).

from noxy_langgraph import NOXY_SENT_DECISION_ID_KEY

class State(TypedDict, total=False):
    ...
    _noxy_sent_decision_id: Optional[str]  # or use NOXY_SENT_DECISION_ID_KEY

Webhook payload

Noxy delivers JSON to your registered webhook URL:

On timeout/expired, pass an on_timeout callback to create_hitl_node to apply default behaviour:

def on_timeout(state, resume):
    return {"noxy_decision": resume.to_state(), "approved": False}

bridge.create_hitl_node(build_actionable, on_timeout=on_timeout)

API

Examples

• examples/basic.py — end-to-end demo with a mock Noxy client
• examples/webhook_server.py — FastAPI server with /runs and /webhooks/noxy

python examples/basic.py

Configure the webhook server:

export NOXY_APP_TOKEN="your-app-token"
export NOXY_IDENTITY_ID="user@example.com"   # or phone, user id, 0x…
uvicorn examples.webhook_server:app --reload

Development

make dev      # editable install with dev + examples extras
make test     # run pytest
make build    # build sdist + wheel
make publish-check  # build and validate with twine

License

MIT