hgateway-sdk 0.2.1

theved.ai HITL Gateway SDK — raise human-in-the-loop interrupts from LangGraph nodes.

hgateway-sdk

Raise structured human-in-the-loop (HITL) interrupts from inside your LangGraph agent and hand the human interaction to the theved.ai HITL Gateway.

A thin client that carries the ask (a typed HitlContent) plus SDK-derived run context, registers it out-of-band with the gateway, and wraps LangGraph's interrupt() so the resumed value comes back as the paired typed response.

What it does / when to use it

Use it whenever a LangGraph node needs a human to approve, decide, edit, or supply context before the graph continues.

The mental model is ownership vs fallback:

• When the gateway accepts an interrupt, it owns the human interaction — routing, notification, TTL/escalation, and collecting the response. Your graph simply suspends and later resumes with the operator's answer.
• When the gateway is unreachable or declines, the SDK falls back to a plain local interrupt(fallback) so your own backend (or test harness) can resolve it.

This lets you adopt the gateway without giving up a working local path.

Install

The SDK is distributed from its (private) GitHub repository — there is no PyPI release. Install it pinned to a release tag:

uv add "hgateway-sdk @ git+https://github.com/theved-ai/hgateway_sdk.git@v0.1.0"
# or with pip:
pip install "hgateway-sdk @ git+https://github.com/theved-ai/hgateway_sdk.git@v0.1.0"

Because the repo is private, the installing environment must be authenticated to GitHub (an SSH key, or a token via GH_TOKEN / a credential helper / a git+https://<token>@github.com/... URL).

Each release tag also has a built wheel + sdist attached as assets on its GitHub Release; you may install directly from a downloaded wheel instead of building from source.

Configure

Settings are resolved from the process environment first; a .env file is consulted only if a required value is still missing. The gateway URL and register endpoint are fixed — every agent talks to the same theved.ai hgateway service, so they are not configurable.

Variable	Required	Default
HGATEWAY_AGENT_API_KEY	yes	—
HGATEWAY_TIMEOUT_SECONDS	no	5

A missing required value raises GatewayBootstrapError.

Quickstart

import hgateway_sdk as hg
from hgateway_sdk import BinaryApprovalContent, Choice, RunFeatures, Routing, Recipients

hg.init()  # reads HGATEWAY_AGENT_API_KEY (or .env); gateway URL/endpoint are fixed

@hg.hitl_node
def review_node(state):
    content = BinaryApprovalContent(
        prompt=f"Approve this action? {state['summary']}",
        choices=[Choice("approve", "Approve"), Choice("reject", "Reject")],
    )
    resp = hg.raise_interrupt(
        "approve-action",
        content,
        features=RunFeatures(routing=Routing(primary=Recipients(channels=["#ops"]))),
        fallback={"decision": "reject"},   # surfaced if the gateway is unreachable/declines
    )
    return {"decision": resp["decision"]}   # BinaryApprovalResponse is a TypedDict: {"decision": str}

Core concepts

Lifecycle

init()  →  @hitl_node  →  raise_interrupt(...)  →  register + pause  →  resume

1. init() once at startup builds and caches a GatewayClient from the env (or an injected transport for tests). raise_interrupt auto-inits if you skip it.
2. @hitl_node decorates the node. It binds the LangGraph state, config, and an occurrence counter into contextvars so the SDK can build the runtime context. Calling raise_interrupt outside such a node raises HitlNodeRequired.
3. raise_interrupt serializes the spec, registers it with the gateway, and suspends the graph.
4. On resume the operator's response is returned as the typed response.

Ownership vs fallback

When the gateway accepts, the SDK suspends with a sentinel value that marks the interrupt as gateway-owned:

{"__hgateway_owned__": True, "hitl_instance_id": "..."}

A backend-for-agent (BFA) inspecting pending interrupts uses this sentinel to know the gateway is driving the interaction and should not render its own UI. When the gateway is unreachable or declines, the SDK instead suspends with your fallback dict (or the full wire spec if you passed none) — that is your local-handling path.

Replay-safety

LangGraph re-executes the node from the top every time the graph resumes, and raise_interrupt re-registers on each replay. Do not place unguarded side effects (charging a card, sending an email, mutating external state) before the interrupt — they will run again on resume. Keep everything before raise_interrupt pure/idempotent, and perform side effects after the response is in hand.

JSON-serializability

Content fields and the graph state are serialized onto the wire. They must be JSON-safe (no dataclass instances, sets, datetimes, etc. nested in state or content values) or registration fails at the transport.

Interrupt families and response shapes

The 4 families map to 10 content types, each paired with a response TypedDict:

Family	Content class	Response shape
Approval	BinaryApprovalContent	{"decision": str}
Approval	ModifyApprovalContent	{"decision": str, "values"?: dict}
Decision	SingleDecisionContent	{"selected": str}
Decision	MultiDecisionContent	{"selected": list[str]}
Decision	RankDecisionContent	{"ranking": list[str]}
Context	FreetextContextContent	{"value": str}
Context	FormContextContent	{"values": dict}
Context	ConfirmContextContent	{"confirmed": bool}
Edit	ContentEditContent	{"content": str}
Edit	ToolArgsEditContent	{"args": dict}

Responses are TypedDicts — access them by key (resp["decision"]), never by attribute.

Recipes

A decision

import hgateway_sdk as hg
from hgateway_sdk import SingleDecisionContent, Choice

resp = hg.raise_interrupt(
    "pick-region",
    SingleDecisionContent(
        prompt="Which region should we deploy to?",
        options=[Choice("us", "US"), Choice("eu", "EU")],
    ),
)
region = resp["selected"]

An approval with routing + TTL

import hgateway_sdk as hg
from hgateway_sdk import BinaryApprovalContent, RunFeatures, Routing, Recipients, Ttl, OnExpiry

features = RunFeatures(
    routing=Routing(
        primary=Recipients(channels=["#ops-approvals"]),
        secondary=Recipients(users=["[e-mail]"]),
    ),
    ttl=Ttl(seconds=1800, on_expiry=OnExpiry.FORWARD_TO_SECONDARY),
)
resp = hg.raise_interrupt("deploy-approval", BinaryApprovalContent(prompt="Deploy?"), features=features)

The fallback pattern

resp = hg.raise_interrupt(
    "approve-action",
    BinaryApprovalContent(prompt="Approve?"),
    fallback={"decision": "reject"},   # used when the gateway is unreachable or declines
)

Testing with a fake transport

Inject a fake GatewayTransport (see tests/e2e_test_suite/fakes.py) so tests never hit the network:

import hgateway_sdk as hg
from tests.e2e_test_suite.fakes import AcceptingTransport
from tests.e2e_test_suite.helpers import build_single_node_graph, run, resume, new_thread
from langgraph.types import Command

hg.init(transport=AcceptingTransport())
# build a 1-node graph around your @hitl_node, run() to the interrupt,
# then resume(graph, {"decision": "approve"}, thread_id).

Error handling

Every exception subclasses HGatewayError, so a single except HGatewayError is a safe catch-all.

Exception	Cause	Remedy
GatewayBootstrapError	The required HGATEWAY_AGENT_API_KEY env var is missing.	Set the env var or provide a .env.
HitlNodeRequired	raise_interrupt called outside an @hitl_node node.	Decorate the calling node with @hg.hitl_node.
FeatureValidationError	A RunFeatures/Routing/Ttl/Recipients is misconfigured (e.g. empty primary).	Fix the feature config; validation runs at construction.
GatewayUnreachable	The HTTP transport could not reach the gateway.	The SDK falls back to a local interrupt(fallback); ensure fallback is set and the gateway is reachable.

Versioning / source

hgateway_sdk.__version__ reports the installed version. Public API is re-exported from the top-level hgateway_sdk namespace. Deep per-symbol reference lives in the inline docstrings (IDE hover / pydoc).