Noxy LangChain Connector

LangChain connector for Noxy human-in-the-loop guardrails. Uses HumanInTheLoopMiddleware to pause agent tool calls, 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 langchain-noxy pulls in noxy-sdk automatically; you do not need a separate checkout of the Noxy SDK.

Flow

sequenceDiagram
    participant A as LangChain Agent
    participant SDK as noxy-sdk
    participant N as Noxy Relay
    participant D as User Devices

    A->>SDK: send_decision (tool approval)
    SDK->>N: RouteDecision
    N->>D: Deliver to registered devices
    A->>A: interrupt() — state saved to checkpointer
    Note over A: Agent suspended

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

    SDK->>A: Command(resume=HITLResponse)
    A->>A: Continue with approved/rejected tool calls

1. The agent proposes a guarded tool call.
2. NoxyHumanInTheLoopMiddleware routes an encrypted actionable to all devices for the identity.
3. LangChain calls interrupt() — the agent 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 agent continues with approved or rejected tool calls.

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

Requirements

• Python >= 3.10
• A LangChain agent created with create_agent and a checkpointer
• A Noxy app token and target identity (phone, email, user id, or wallet 0x…)

Installation

pip install langchain-noxy

Optional FastAPI example server:

pip install "langchain-noxy[examples]"

Configuration

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

Copy .env.example to .env when running 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 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."""
    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")

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 when transfer_funds is proposed; read decision_id from interrupt

final = bridge.wait_and_resume(agent, "<decision_id>")

Manual polling

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

Outcome mapping

Noxy relay outcomes map to LangChain HITL decisions (via hitl_response_from_outcome):

• approved → {"type": "approve"}
• rejected / expired / timeout → {"type": "reject", "message": "..."}

Poll tuning

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

API

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/langchain-connector.git
cd langchain-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="user@example.com"
uvicorn examples.poll_resume_server:app --reload

Development

For contributors working on this repository:

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

License

MIT