noxy-langgraph 1.0.0

LangGraph connector for Noxy human-in-the-loop decisions via interrupt/resume and webhooks.

Noxy LangGraph Connector

LangGraph connector for Noxy human-in-the-loop guardrails. Pauses agent graphs with interrupt(), delivers encrypted approval prompts to user devices, and resumes execution when Noxy fires a webhook.

Flow

sequenceDiagram
    participant G as LangGraph
    participant N as Noxy Relay
    participant P as User Phone
    participant S as Your Server

    G->>N: send_decision (push payload)
    N->>P: Push notification
    G->>G: interrupt() — state saved to checkpointer
    Note over G: Graph suspended

    alt User responds
        P->>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 the user's device (push).
3. The node calls interrupt() — LangGraph suspends and persists state via a checkpointer.
4. User responds on mobile 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, "[e-mail]")

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": "[e-mail]",
    "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-send the push 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:

Field	Type	Description
decisionId	str	Decision to resume (snake_case decision_id also accepted)
identityId	str	Identity that took the decision (phone, email, user id, or wallet)
outcome	str	approved, rejected, expired, or timeout
receivedAt	str	Optional ISO timestamp

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

Symbol	Description
NoxyLangGraphBridge	Wires client, registry, HITL node factory, and resume handler
create_noxy_hitl_node(...)	Low-level HITL node factory
NoxyGraphResumeHandler	Resume paused graphs from webhook payloads
PendingInterruptRegistry	Maps decision_id → thread_id for resume
build_tool_call_actionable(...)	Standard propose_tool_call payload builder
parse_webhook_payload(...)	Parse raw webhook JSON

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="[e-mail]"   # 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

Publishing

1. Bump version in pyproject.toml and noxy_langgraph/_version.py.
2. Update CHANGELOG.md.
3. Merge to main — the GitHub Actions publish workflow uploads to PyPI when PYPI_API_TOKEN is configured.

License

MIT