Client & Decorators

The entry points you call directly: initialize once, decorate your node, raise the interrupt. See the Integration Guide for the order these are actually used in.

hgateway_sdk.GatewayClient

GatewayClient(transport: GatewayTransport, *, agent_api_key: str)

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

Args: transport: Transport implementation used to send the interrupt spec to the gateway. agent_api_key: Bearer token sent with every request to authenticate the agent with the gateway.

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/client/gateway_client.py

def __init__(self, transport: GatewayTransport, *, agent_api_key: str):
    self._transport = transport
    self._agent_api_key = agent_api_key

raise_interrupt

raise_interrupt(hitl_key: str, content: HitlContent[TResp], features: Optional[RunFeatures] = None, *, fallback: Optional[dict] = None) -> TResp

Register an interrupt with the gateway and suspend the LangGraph node.

Builds the interrupt envelope (:class:~hgateway_sdk.HitlSpec) from the supplied arguments and the runtime context captured by the @hg.hitl_node decorator, serialises it to a wire dict, and attempts to register it with the gateway.

Ownership decision

• If the transport call succeeds and ack.accepted is True, the gateway owns the interrupt. The method still raises a LangGraph interrupt (so the graph checkpoints and pauses for the gateway to resume) but surfaces a sentinel value — {"__hgateway_owned__": True, "hitl_instance_id": ...} — rather than bare None, so a backend consuming the stream can detect and ignore gateway-owned interrupts. The operator's response is delivered on resume regardless of the surfaced value.
• If :class:~hgateway_sdk.GatewayUnreachable is raised or the gateway returns accepted=False, the method falls back to a local interrupt(fallback or wire) so the agent's own backend handles it.

Args: hitl_key: Logical name for this interrupt type, e.g. "approve-deployment". Combined with the occurrence counter to form a dedup ID. content: Typed content describing the review task (a subclass of :class:~hgateway_sdk.HitlContent). features: Optional :class:~hgateway_sdk.RunFeatures (routing, TTL, recipients, cross-feature rules). Defaults to an empty RunFeatures() if None. fallback: Raw dict passed to interrupt() when the gateway does not take ownership. If None, the serialised wire spec is used.

Returns: The typed operator response on graph resume.

Raises: HitlNodeRequired: If the method is called outside a node decorated with @hg.hitl_node (the runtime contextvars are unset).

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/client/gateway_client.py

def raise_interrupt(
    self,
    hitl_key: str,
    content: HitlContent[TResp],
    features: Optional[RunFeatures] = None,
    *,
    fallback: Optional[dict] = None,
) -> TResp:
    """Register an interrupt with the gateway and suspend the LangGraph node.

    Builds the interrupt envelope (:class:`~hgateway_sdk.HitlSpec`) from
    the supplied arguments and the runtime context captured by the
    ``@hg.hitl_node`` decorator, serialises it to a wire dict, and
    attempts to register it with the gateway.

    **Ownership decision**

    - If the transport call succeeds and ``ack.accepted`` is ``True``, the
      gateway owns the interrupt.  The method still raises a LangGraph
      interrupt (so the graph checkpoints and pauses for the gateway to
      resume) but surfaces a sentinel value —
      ``{"__hgateway_owned__": True, "hitl_instance_id": ...}`` — rather than
      bare ``None``, so a backend consuming the stream can detect and
      ignore gateway-owned interrupts.  The operator's response is
      delivered on resume regardless of the surfaced value.
    - If :class:`~hgateway_sdk.GatewayUnreachable` is raised or the
      gateway returns ``accepted=False``, the method falls back to a local
      ``interrupt(fallback or wire)`` so the agent's own backend handles it.

    Args:
        hitl_key: Logical name for this interrupt type, e.g.
            ``"approve-deployment"``.  Combined with the occurrence counter
            to form a dedup ID.
        content: Typed content describing the review task (a subclass of
            :class:`~hgateway_sdk.HitlContent`).
        features: Optional :class:`~hgateway_sdk.RunFeatures` (routing,
            TTL, recipients, cross-feature rules).  Defaults to an empty
            ``RunFeatures()`` if ``None``.
        fallback: Raw dict passed to ``interrupt()`` when the gateway does
            not take ownership.  If ``None``, the serialised wire spec is
            used.

    Returns:
        The typed operator response on graph resume.

    Raises:
        HitlNodeRequired: If the method is called outside a node decorated
            with ``@hg.hitl_node`` (the runtime contextvars are unset).
    """
    runtime = derive_runtime(next_occurrence(hitl_key))
    spec = HitlSpec(
        hitl_key=hitl_key,
        content=content,
        runtime=runtime,
        features=features or RunFeatures(),
    )
    wire = spec_to_wire(spec)

    ack = None
    try:
        ack = self._transport.register(wire, agent_api_key=self._agent_api_key)
        gateway_owns = ack.accepted
    except GatewayUnreachable:
        gateway_owns = False

    if gateway_owns:
        # The gateway owns this interrupt, but we still raise a LangGraph
        # interrupt so the graph checkpoints and pauses for the gateway to
        # resume. Surface a sentinel (not bare None) so a backend consuming
        # the stream can positively identify gateway-owned interrupts and
        # ignore them. The return value on resume is the gateway-supplied
        # `Command(resume=...)` payload, unaffected by what we pass here.
        return interrupt(
            {
                constants.GATEWAY_OWNED_MARKER: True,
                constants.ACK_HITL_INSTANCE_ID: ack.hitl_instance_id if ack else None,
            }
        )
    # Gateway didn't take ownership (unreachable or declined): raise locally so
    # the agent's own backend handles it. `fallback` lets the caller supply the
    # exact raw payload their BFA expects; otherwise we raise the SDK wire.
    return interrupt(fallback if fallback is not None else wire)

hgateway_sdk.init

init(transport: Optional[GatewayTransport] = None) -> GatewayClient

Initialise the SDK and return the active :class:GatewayClient.

Reads the HGATEWAY_AGENT_API_KEY environment variable (or a .env file via python-dotenv); the gateway URL and register endpoint are fixed, not read from the environment. Call this once at application startup — typically in your agent's main or lifespan hook — before any node runs.

If transport is omitted, an :class:~hgateway_sdk.HttpGatewayTransport is constructed from the loaded settings. Pass a custom transport to inject a test double or an alternative HTTP implementation.

The returned client is stored in a module-level singleton; subsequent calls to :func:raise_interrupt use it automatically. Calling init a second time replaces the singleton.

Args: transport: Optional custom transport implementing :class:~hgateway_sdk.GatewayTransport. If None, the default HTTP transport is used.

Returns: The newly created and cached :class:GatewayClient.

Raises: GatewayBootstrapError: If required environment variables are missing and cannot be resolved from a .env file.

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/client/entrypoint.py

def init(transport: Optional[GatewayTransport] = None) -> GatewayClient:
    """Initialise the SDK and return the active :class:`GatewayClient`.

    Reads the ``HGATEWAY_AGENT_API_KEY`` environment variable (or a ``.env``
    file via python-dotenv); the gateway URL and register endpoint are fixed,
    not read from the environment.  Call this once at application startup —
    typically in your agent's ``main`` or ``lifespan`` hook — before any node
    runs.

    If ``transport`` is omitted, an :class:`~hgateway_sdk.HttpGatewayTransport`
    is constructed from the loaded settings.  Pass a custom transport to inject
    a test double or an alternative HTTP implementation.

    The returned client is stored in a module-level singleton; subsequent calls
    to :func:`raise_interrupt` use it automatically.  Calling ``init`` a second
    time replaces the singleton.

    Args:
        transport: Optional custom transport implementing
            :class:`~hgateway_sdk.GatewayTransport`.  If ``None``, the default
            HTTP transport is used.

    Returns:
        The newly created and cached :class:`GatewayClient`.

    Raises:
        GatewayBootstrapError: If required environment variables are missing
            and cannot be resolved from a ``.env`` file.
    """
    global _client
    settings = load_settings()
    resolved_transport = transport or HttpGatewayTransport(
        settings.url, settings.register_endpoint, settings.timeout
    )
    _client = GatewayClient(resolved_transport, agent_api_key=settings.agent_api_key)
    return _client

hgateway_sdk.raise_interrupt

raise_interrupt(hitl_key: str, content: HitlContent[TResp], features: Optional[RunFeatures] = None, *, fallback: Optional[dict] = None) -> TResp

Register a HITL interrupt with the gateway and pause the graph.

Convenience wrapper around :meth:GatewayClient.raise_interrupt. Must be called from inside a node decorated with :func:~hgateway_sdk.hitl_node; the decorator populates the contextvars that this call reads to build the runtime context (thread ID, run ID, occurrence counter).

Gateway-ownership vs fallback behaviour

1. The interrupt spec is serialised and sent to the gateway via the configured transport.
2. If the gateway responds with accepted=True, the SDK calls langgraph.types.interrupt(None) — the graph suspends and control returns to the LangGraph server, which will resume the node with the operator's response once the gateway notifies it.
3. If the gateway is unreachable (:class:~hgateway_sdk.GatewayUnreachable) or returns accepted=False, the SDK falls back to a local interrupt: interrupt(fallback) if fallback is provided, otherwise interrupt(wire) where wire is the full serialised spec dict.

Must be called inside a @hg.hitl_node-decorated node.

Example::

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}

Args: hitl_key: Logical, stable identifier for this interrupt type, e.g. "approve-sql-query". Used as part of dedup-ID generation (with checkpoint_ns and occurrence). Treat it as immutable: use a fixed literal per interrupt site and never derive it from per-run values (UUIDs, timestamps, state). The same logical interrupt must pass the same key on every (re)run, because the node is replayed on resume and a changed key reads as a different interrupt — breaking dedup and any in-flight gateway interaction. See :class:~hgateway_sdk.HitlSpec for the full rationale. content: Typed content object (a :class:~hgateway_sdk.HitlContent subclass) describing what the operator needs to review. features: Optional :class:~hgateway_sdk.RunFeatures controlling routing, TTL, recipients, and other gateway features. fallback: Raw dict passed to interrupt() when the gateway does not take ownership. If None, the full wire-format spec dict is used instead.

Returns: The typed response object from the operator (a subclass of the response type parameterised by content), delivered on graph resume.

Raises: HitlNodeRequired: If called outside a @hg.hitl_node-decorated node (the runtime contextvars are unset).

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/client/entrypoint.py

def raise_interrupt(
    hitl_key: str,
    content: HitlContent[TResp],
    features: Optional[RunFeatures] = None,
    *,
    fallback: Optional[dict] = None,
) -> TResp:
    """Register a HITL interrupt with the gateway and pause the graph.

    Convenience wrapper around :meth:`GatewayClient.raise_interrupt`.  Must be
    called from inside a node decorated with :func:`~hgateway_sdk.hitl_node`;
    the decorator populates the contextvars that this call reads to build the
    runtime context (thread ID, run ID, occurrence counter).

    **Gateway-ownership vs fallback behaviour**

    1. The interrupt spec is serialised and sent to the gateway via the
       configured transport.
    2. If the gateway responds with ``accepted=True``, the SDK calls
       ``langgraph.types.interrupt(None)`` — the graph suspends and control
       returns to the LangGraph server, which will resume the node with the
       operator's response once the gateway notifies it.
    3. If the gateway is unreachable (:class:`~hgateway_sdk.GatewayUnreachable`)
       or returns ``accepted=False``, the SDK falls back to a local interrupt:
       ``interrupt(fallback)`` if ``fallback`` is provided, otherwise
       ``interrupt(wire)`` where ``wire`` is the full serialised spec dict.

    Must be called inside a ``@hg.hitl_node``-decorated node.

    Example::

        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}

    Args:
        hitl_key: Logical, stable identifier for this interrupt type, e.g.
            ``"approve-sql-query"``.  Used as part of dedup-ID generation
            (with ``checkpoint_ns`` and ``occurrence``).  **Treat it as
            immutable:** use a fixed literal per interrupt site and never derive
            it from per-run values (UUIDs, timestamps, state).  The same logical
            interrupt must pass the same key on every (re)run, because the node
            is replayed on resume and a changed key reads as a different
            interrupt — breaking dedup and any in-flight gateway interaction.
            See :class:`~hgateway_sdk.HitlSpec` for the full rationale.
        content: Typed content object (a :class:`~hgateway_sdk.HitlContent`
            subclass) describing what the operator needs to review.
        features: Optional :class:`~hgateway_sdk.RunFeatures` controlling
            routing, TTL, recipients, and other gateway features.
        fallback: Raw dict passed to ``interrupt()`` when the gateway does not
            take ownership.  If ``None``, the full wire-format spec dict is
            used instead.

    Returns:
        The typed response object from the operator (a subclass of the response
        type parameterised by ``content``), delivered on graph resume.

    Raises:
        HitlNodeRequired: If called outside a ``@hg.hitl_node``-decorated node
            (the runtime contextvars are unset).
    """
    return _ensure_client().raise_interrupt(
        hitl_key, content, features, fallback=fallback
    )

hgateway_sdk.hitl_node

hitl_node(fn)

Decorator that makes a LangGraph node HITL-aware.

Wraps a node function so that, before the body executes, the current graph state and LangGraph config are bound into SDK contextvars via :func:~hgateway_sdk.runtime.bind. This binding is required by :func:~hgateway_sdk.raise_interrupt, which reads those vars to populate the :class:~hgateway_sdk.RuntimeContext (thread ID, run ID, occurrence counter).

The decorator is transparent to LangGraph: it preserves the function name, signature, and return value. Both synchronous and async def nodes are supported — an async node is wrapped in an async wrapper so the contextvar binding stays active across await points within the node's task. The config argument is optional — if the wrapped function does not declare it, the decorator still fetches the active config via langgraph.config.get_config and uses it for the runtime context without passing it to fn.

Calling :func:~hgateway_sdk.raise_interrupt outside a @hitl_node- decorated node raises :class:~hgateway_sdk.HitlNodeRequired.

Usage::

import hgateway_sdk as hg

@hg.hitl_node
def my_review_node(state):
    resp = hg.raise_interrupt("review", MyContent(...))
    return {"result": resp}

Args: fn: The node function to wrap. Must accept state as its first positional argument; may optionally accept config as a second argument.

Returns: A wrapped function with identical semantics, safe to register as a LangGraph node.

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/decorators.py

def hitl_node(fn):
    """Decorator that makes a LangGraph node HITL-aware.

    Wraps a node function so that, before the body executes, the current graph
    ``state`` and LangGraph ``config`` are bound into SDK contextvars via
    :func:`~hgateway_sdk.runtime.bind`.  This binding is required by
    :func:`~hgateway_sdk.raise_interrupt`, which reads those vars to populate
    the :class:`~hgateway_sdk.RuntimeContext` (thread ID, run ID, occurrence
    counter).

    The decorator is transparent to LangGraph: it preserves the function name,
    signature, and return value.  Both synchronous and ``async def`` nodes are
    supported — an async node is wrapped in an async wrapper so the contextvar
    binding stays active across ``await`` points within the node's task.  The
    ``config`` argument is optional — if the wrapped function does not declare
    it, the decorator still fetches the active config via
    ``langgraph.config.get_config`` and uses it for the runtime context without
    passing it to ``fn``.

    Calling :func:`~hgateway_sdk.raise_interrupt` outside a ``@hitl_node``-
    decorated node raises :class:`~hgateway_sdk.HitlNodeRequired`.

    Usage::

        import hgateway_sdk as hg

        @hg.hitl_node
        def my_review_node(state):
            resp = hg.raise_interrupt("review", MyContent(...))
            return {"result": resp}

    Args:
        fn: The node function to wrap.  Must accept ``state`` as its first
            positional argument; may optionally accept ``config`` as a second
            argument.

    Returns:
        A wrapped function with identical semantics, safe to register as a
        LangGraph node.
    """
    wants_config = "config" in inspect.signature(fn).parameters

    def _resolve(config):
        if config is None:
            with suppress(RuntimeError):
                config = get_config()
        return config

    if inspect.iscoroutinefunction(fn):

        @functools.wraps(fn)
        async def async_wrapper(state, config=None):
            tokens = runtime.bind(state, _resolve(config))
            try:
                return await (fn(state, config) if wants_config else fn(state))
            finally:
                runtime.unbind(tokens)

        return async_wrapper

    @functools.wraps(fn)
    def wrapper(state, config=None):
        tokens = runtime.bind(state, _resolve(config))
        try:
            return fn(state, config) if wants_config else fn(state)
        finally:
            runtime.unbind(tokens)

    return wrapper

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.

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

Returns: An :class:~hgateway_sdk.Ack indicating whether the gateway accepted ownership of the interrupt.

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

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/transport/gateway_transport.py

def register(self, wire: dict, *, agent_api_key: str) -> Ack:
    """Send the serialised interrupt spec to the gateway.

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

    Returns:
        An :class:`~hgateway_sdk.Ack` indicating whether the gateway
        accepted ownership of the interrupt.

    Raises:
        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.

Args: base_url: Root URL of the gateway service (trailing slash stripped). register_endpoint: Path to the interrupt-registration endpoint (leading slash stripped and appended to base_url). timeout: Socket timeout in seconds for the HTTP request.

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/transport/http_gateway_transport.py

def __init__(self, base_url: str, register_endpoint: str, timeout: float):
    self._endpoint = base_url.rstrip("/") + "/" + register_endpoint.lstrip("/")
    self._timeout = timeout

register

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

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

Args: wire: Serialised interrupt envelope dict. agent_api_key: Bearer token sent in the Authorization header.

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

Raises: 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.

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/transport/http_gateway_transport.py

def register(self, wire: dict, *, agent_api_key: str) -> Ack:
    """POST the interrupt spec to the gateway and return the acknowledgement.

    Args:
        wire: Serialised interrupt envelope dict.
        agent_api_key: Bearer token sent in the ``Authorization`` header.

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

    Raises:
        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.
    """
    body = json.dumps(wire).encode(constants.ENCODING_UTF8)
    request = urllib.request.Request(
        self._endpoint,
        data=body,
        method=constants.HTTP_METHOD_POST,
        headers={
            constants.HEADER_CONTENT_TYPE: constants.CONTENT_TYPE_JSON,
            constants.HEADER_AUTHORIZATION: f"{constants.BEARER_SCHEME} {agent_api_key}",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=self._timeout) as response:
            try:
                payload = _read_json(response)
            except json.JSONDecodeError as exc:
                raise GatewayUnreachable(
                    f"gateway returned a non-JSON response: {exc}"
                ) from exc
            status = payload.get(constants.ACK_STATUS)
            # Success body is {"hitl_instance_id": ..., "status": "raised"}.
            # Treat the interrupt as accepted (gateway-owned) on HTTP 2xx,
            # and — when the gateway reports a status — only when it is
            # "raised". A missing status falls back to the HTTP code so the
            # SDK still works against gateways that omit it.
            accepted = 200 <= response.status < 300 and status in (
                None,
                constants.STATUS_RAISED,
            )
            return Ack(
                accepted=accepted,
                hitl_instance_id=payload.get(constants.ACK_HITL_INSTANCE_ID),
                status=status,
                raw=payload,
            )
    except urllib.error.HTTPError as exc:
        # Error body shape: {code, message, request_id, details: []}.
        # Preserve it on the Ack for diagnostics; the gateway declined, so
        # the SDK falls back to local handling (accepted=False).
        return Ack(accepted=False, raw=_read_error_body(exc))
    except urllib.error.URLError as exc:
        raise GatewayUnreachable(str(exc.reason)) from exc

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: thread_id: LangGraph thread identifier (from config["configurable"]["thread_id"]). The gateway uses this to correlate the interrupt with the correct graph run. run_id: 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: 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: 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: 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.