langgraph-noxy 1.0.2

LangGraph connector for Noxy human-in-the-loop.

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 you poll relay for the settled outcome via noxy-sdk (GetDecisionOutcome).

Installing langgraph-noxy pulls in noxy-sdk automatically; you do not need a separate checkout of the Noxy SDK.

Flow

sequenceDiagram
    participant G as LangGraph
    participant SDK as noxy-sdk
    participant N as Noxy Relay
    participant D as User Devices

    G->>SDK: send_decision
    SDK->>N: RouteDecision
    N->>D: Deliver to registered devices
    G->>G: interrupt() — state saved to checkpointer
    Note over G: Graph suspended

    loop Poll GetDecisionOutcome
        SDK->>N: get_decision_outcome
        N-->>SDK: pending / approved / rejected / expired
    end

    SDK->>G: Command(resume=outcome)
    G->>G: Continue with decision in state

1. Graph reaches the HITL node.
2. Noxy routes an encrypted actionable to all devices registered for the identity.
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 on relay.
5. Your process calls wait_for_decision_outcome (SDK) or bridge.wait_and_resume(...).
6. The graph continues with the human decision (or timeout default) in state.

Relay delivers outcomes via gRPC polling (GetDecisionOutcome), with exponential backoff in the SDK.

Requirements

• Python >= 3.10
• A LangGraph graph compiled with a checkpointer (required for interrupt())
• A Noxy app token and target identity (phone, email, user id, or wallet 0x…)

Installation

pip install langgraph-noxy

Optional FastAPI example server:

pip install "langgraph-noxy[examples]"

Configuration

Set credentials in your environment or pass them to NoxyConfig:

Variable	Required	Default	Description
NOXY_APP_TOKEN	Yes	—	App token from the Noxy dashboard (Bearer auth to relay)
NOXY_IDENTITY_ID	Yes*	—	Target identity: phone, email, user id, or wallet address
NOXY_ENDPOINT	No	https://relay.noxy.network	Relay gRPC endpoint

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

Copy .env.example to .env when running the repository examples locally (never commit real tokens).

import os
from noxy import NoxyConfig, init_noxy_agent_client

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

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

Manual polling

If you already poll elsewhere, resume from a single get_decision_outcome response:

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

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.

from langgraph_noxy import NOXY_SENT_DECISION_ID_KEY

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

Poll tuning

Pass WaitForDecisionOutcomeOptions to bridge.wait_and_resume (same fields as noxy-sdk):

Field	Default	Description
initial_poll_interval_ms	400	First delay between polls
max_poll_interval_ms	30000	Cap between polls
max_wait_ms	900000	Stop polling and resume with timeout outcome
backoff_multiplier	1.6	Exponential backoff factor

On timeout (poll budget exceeded), pass an on_timeout callback to create_hitl_node:

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, and wait_and_resume
create_noxy_hitl_node(...)	Low-level HITL node factory
NoxyGraphResumeHandler.wait_and_resume	SDK poll loop + Command(resume=...)
NoxyGraphResumeHandler.resume_from_poll_response	Resume from one terminal poll
PendingInterruptRegistry	Maps decision_id → thread_id for resume
build_tool_call_actionable(...)	Standard propose_tool_call payload builder
parse_webhook_payload(...)	Optional: parse webhook-shaped JSON if you bridge events yourself

Examples

Example scripts are maintained in the GitHub repository (they are not shipped inside the PyPI wheel). Clone the repo to run them:

git clone https://github.com/noxy-network/langgraph-connector.git
cd langgraph-connector
pip install ".[examples]"
cp .env.example .env   # set NOXY_APP_TOKEN and NOXY_IDENTITY_ID

• examples/basic.py — mock client, no relay required
• examples/poll_resume_server.py — FastAPI: POST /runs, then POST /runs/wait

python examples/basic.py

export NOXY_APP_TOKEN="your-app-token"
export NOXY_IDENTITY_ID="[e-mail]"
uvicorn examples.poll_resume_server:app --reload

Development

For contributors working on this repository:

git clone https://github.com/noxy-network/langgraph-connector.git
cd langgraph-connector
pip install ".[dev,examples]"
make test
make build
make publish-check

License

MIT