Client & Decorators

The entry points you call directly, in the order you actually use them. See the Integration Guide for the full walkthrough.

Entry points

init() isn't required before use — raise_interrupt auto-initializes the client from HGATEWAY_AGENT_API_KEY on first call. Call it yourself only if you need to inject a custom transport (e.g. a fake for tests) or to set default routing/ttl for the whole agent.

init, hitl_node, and raise_interrupt all accept the same feature kwargs (channels, users, secondary_channels, secondary_users, ttl_seconds, on_expiry, default_response, or a full features=RunFeatures(...)). They form a 3-scope cascade — init is the lowest-priority default, hitl_node overrides it per node, and the raise_interrupt/raise_* call site overrides both. Routing and Ttl are mandatory: a HITL cannot be raised unless both resolve to a concrete value from some scope. Use CLEAR_FEATURE to explicitly unset a value inherited from a lower scope. See Features & Routing for the full cascade reference.

Decorate a node with hitl_node before calling raise_interrupt inside it — it binds the current LangGraph state, config, and this node's feature defaults into contextvars so the SDK can build the runtime context and resolve the cascade automatically.

raise_interrupt is the call that actually registers a HITL with the gateway and suspends the node. Must be called from inside a @hitl_node-decorated node. For the common content types, prefer the raise_* wrapper functions (raise_approval, raise_pick_one_option, ...) documented in Wrapper Functions — they build the content object for you.

hgateway_sdk.init

init(transport: Optional[GatewayTransport] = None, *, channels: ChannelsInput = None, users: UsersInput = None, secondary_channels: ChannelsInput = None, secondary_users: UsersInput = None, ttl_seconds: TtlSecondsInput = None, on_expiry: OnExpiryInput = None, default_response: DefaultResponseInput = None, features: Optional[RunFeatures] = None) -> GatewayClient

Initialise the SDK and return the active GatewayClient.

Reads HGATEWAY_AGENT_API_KEY (env var or .env); the gateway URL/endpoint are fixed, not configurable. Call once at startup, before any node runs — calling it again replaces the cached client every raise_interrupt uses. channels/users/ttl_seconds/on_expiry/etc. set here are the cascade's lowest-priority scope, overridable per hitl_node or per call site (see resolve_features). Routing and ttl must each resolve to a value from some scope before any HITL can be raised — they don't have to be set here.

Example::

import hgateway_sdk as hg

hg.init(channels=["#ops"], ttl_seconds=3600, on_expiry="default-response")

hgateway_sdk.hitl_node

hitl_node(fn=None, *, channels: ChannelsInput = None, users: UsersInput = None, secondary_channels: ChannelsInput = None, secondary_users: UsersInput = None, ttl_seconds: TtlSecondsInput = None, on_expiry: OnExpiryInput = None, default_response: DefaultResponseInput = None, features: Optional[RunFeatures] = None)

Bind graph state/config into SDK contextvars for the wrapped node.

Apply to the exact function or method that calls raise_interrupt — not to a class, since the decorator must see the specific callable LangGraph invokes to know which positional arg is state. Feature kwargs set here apply to every raise_interrupt call inside this node, overridable per call site (see resolve_features).

hgateway_sdk.raise_interrupt

raise_interrupt(hitl_key: str, content: HitlContent[TResp], features: Optional[RunFeatures] = None, *, fallback: Optional[dict] = None, channels: ChannelsInput = None, users: UsersInput = None, secondary_channels: ChannelsInput = None, secondary_users: UsersInput = None, ttl_seconds: TtlSecondsInput = None, on_expiry: OnExpiryInput = None, default_response: DefaultResponseInput = None) -> TResp

Raise a human-in-the-loop interrupt and pause the graph for a response.

hitl_key identifies this call site (not a single run) — every HITL raised from this exact spot, across every run, shares this key; never derive it from per-run values. If the gateway is unreachable, fallback (if given) takes over locally instead of the HITL just hanging. Must be called inside a @hg.hitl_node-decorated node. channels/users/ttl_seconds/on_expiry/etc. override this call's inherited cascade values (see resolve_features); for the common content types, prefer the hitl_wrapper functions (raise_approval, raise_pick_one_option, ...) over constructing content directly.

Example::

import hgateway_sdk as hg
from hgateway_sdk import BinaryApprovalContent, Choice

hg.init(channels=["#ops"], ttl_seconds=3600, on_expiry="default-response")

@hg.hitl_node
def review_node(state):
    resp = hg.raise_interrupt(
        "approve-action",
        BinaryApprovalContent(prompt="Approve?", choices=[Choice("approve", "Approve")]),
        fallback={"decision": "reject"},
    )
    return {"decision": resp["decision"]}

Raises:

Type	Description
HitlNodeRequired	If called outside a @hg.hitl_node-decorated node.
FeatureValidationError	If routing or ttl has no value from any cascade scope.

Advanced

You won't normally touch these directly — raise_interrupt uses them under the hood. Documented here for anyone injecting a custom transport or reading the runtime context by hand.

GatewayClient is the stateful object init() builds and raise_interrupt delegates to. Its own raise_interrupt method is the same call documented above under Entry points — not repeated here.

HttpGatewayTransport is the default transport, used unless you pass a custom one to init().

hgateway_sdk.GatewayClient

GatewayClient(transport: GatewayTransport, *, agent_api_key: str, default_feature_kwargs: Optional[dict] = None)

Stateful client that registers HITL interrupts with the hgateway service.

Instantiated by :func:~hgateway_sdk.init and cached as a module-level singleton in entrypoint. You generally interact with this class indirectly through :func:~hgateway_sdk.raise_interrupt, but you can obtain the client instance directly from init() if you prefer explicit dependency injection.

The client is transport-agnostic: pass any object implementing the :class:~hgateway_sdk.GatewayTransport protocol to replace the default HTTP transport (useful for testing).

Parameters:

Name	Type	Description	Default
transport	GatewayTransport	Transport implementation used to send the interrupt spec to the gateway.	required
agent_api_key	str	Bearer token sent with every request to authenticate the agent with the gateway.	required

hgateway_sdk.GatewayTransport

Bases: Protocol

Protocol for sending a HITL interrupt spec to the gateway.

Any object that implements register with this exact signature satisfies the protocol and can be passed as the transport argument to :func:~hgateway_sdk.init or :class:~hgateway_sdk.GatewayClient.

Use this protocol to inject a custom transport — for example, a test double that returns a controlled :class:~hgateway_sdk.Ack without making real HTTP calls::

class AlwaysAcceptTransport:
    def register(self, wire: dict, *, agent_api_key: str) -> Ack:
        return Ack(accepted=True, hitl_instance_id="test-id", raw={})

client = hg.init(transport=AlwaysAcceptTransport())

The protocol is @runtime_checkable, so isinstance(obj, GatewayTransport) works for basic structural verification.

register

register(wire: dict, *, agent_api_key: str) -> Ack

Send the serialised interrupt spec to the gateway.

Parameters:

Name	Type	Description	Default
wire	dict	The fully serialised interrupt envelope as a plain dict (produced by :func:~hgateway_sdk.utils.serialization.spec_to_wire).	required
agent_api_key	str	Bearer token for authenticating the agent with the gateway service.	required

Returns:

Name	Type	Description
An	Ack	class:~hgateway_sdk.Ack indicating whether the gateway
	Ack	accepted ownership of the interrupt.

Raises:

Type	Description
GatewayUnreachable	If the gateway cannot be reached (network error, timeout, DNS failure, etc.).

hgateway_sdk.HttpGatewayTransport

HttpGatewayTransport(base_url: str, register_endpoint: str, timeout: float)

Default transport that registers interrupts over HTTPS/JSON.

Constructed automatically by :func:~hgateway_sdk.init from settings resolved by :func:~hgateway_sdk.config.load_settings. You can also construct it directly if you need fine-grained control over the endpoint or timeout::

transport = HttpGatewayTransport(
    base_url="https://gateway.example.com",
    register_endpoint="/api/v1/sdk/hitl/register",
    timeout=5.0,
)
client = hg.init(transport=transport)

A HTTP 2xx response with body {"hitl_instance_id": ..., "status": "raised"} is treated as accepted=True (gateway-owned). HTTP errors are treated as accepted=False (the gateway declined) and their structured error body (code/message/request_id/details) is preserved on Ack.raw; network/DNS failures and non-JSON 2xx bodies raise :class:~hgateway_sdk.GatewayUnreachable.

Parameters:

Name	Type	Description	Default
base_url	str	Root URL of the gateway service (trailing slash stripped).	required
register_endpoint	str	Path to the interrupt-registration endpoint (leading slash stripped and appended to base_url).	required
timeout	float	Socket timeout in seconds for the HTTP request.	required

register

register(wire: dict, *, agent_api_key: str) -> Ack

POST the interrupt spec to the gateway and return the acknowledgement.

Parameters:

Name	Type	Description	Default
wire	dict	Serialised interrupt envelope dict.	required
agent_api_key	str	Bearer token sent in the Authorization header.	required

Returns:

Type	Description
Ack	class:~hgateway_sdk.Ack with accepted=True for HTTP 2xx,
Ack	accepted=False for HTTP errors.

Raises:

Type	Description
GatewayUnreachable	For network-level failures (URLError, timeouts, DNS errors) and for a 2xx response whose body is not valid JSON — an unintelligible gateway response means ownership cannot be confirmed, so the SDK fails safe to its fallback path rather than pausing indefinitely on a malfunctioning gateway.

hgateway_sdk.RuntimeContext dataclass

RuntimeContext(thread_id: str, run_id: str, occurrence: int, checkpoint_ns: str = '', state: dict = dict())

LangGraph runtime identifiers and graph state for a single interrupt.

Automatically populated by the SDK decorator from the LangGraph RunnableConfig and interrupt counter. Developers do not create this manually.

Attributes:

Name	Type	Description
thread_id	str	LangGraph thread identifier (from config["configurable"]["thread_id"]). The gateway uses this to correlate the interrupt with the correct graph run.
run_id	str	LangGraph run identifier (from config["run_id"] / config["configurable"]["run_id"]). Note this changes on every resume (each invoke is a new run), so it is not part of the dedup identity.
checkpoint_ns	str	LangGraph per-task checkpoint namespace (from config["configurable"]["checkpoint_ns"], formatted "<node>:<task-uuid>"). It is stable when the same interrupt is replayed on resume but distinct on each graph-cycle re-entry, so together with hitl_key and occurrence it gives the gateway a dedup identity that is replay-stable yet unique per loop iteration. Defaults to an empty string when unavailable.
occurrence	int	0-based counter tracking how many times this particular hitl_key has been raised within the current node execution. Disambiguates multiple raises of the same name inside one node body; combined with checkpoint_ns for cross-cycle uniqueness.
state	dict	Snapshot of the LangGraph graph state at the moment the interrupt is raised. Forwarded to the gateway so operators and routing rules can inspect graph context. Defaults to an empty dict.