Features & Routing

Pass a RunFeatures to raise_interrupt to control who gets notified (Routing/Recipients) and what happens if nobody responds in time (Ttl/OnExpiry).

hgateway_sdk.RunFeatures dataclass

RunFeatures(routing: Optional[Routing] = None, ttl: Optional[Ttl] = None)

Container for optional per-interrupt feature overrides.

Pass an instance of this class as the features argument to raise_interrupt() to control how the gateway handles the interrupt (routing, deadline, escalation, etc.). __post_init__ runs validate_features immediately so mis-configurations are caught at construction time, not at wire time.

Usage example::

import hgateway_sdk as hg

features = hg.RunFeatures(
    routing=hg.Routing(
        primary=hg.Recipients(channels=["#ops-approvals"]),
    ),
    ttl=hg.Ttl(seconds=1800, on_expiry="default-response",
                default_response={"decision": "reject"}),
)
hg.raise_interrupt("deploy_approval", content, features=features)

Attributes: routing: Optional Routing feature specifying primary (and optional secondary) notification targets. Defaults to None (no explicit routing; the gateway uses its own default). ttl: Optional Ttl feature enforcing a response deadline. Defaults to None (no deadline).

hgateway_sdk.Routing dataclass

Routing(primary: Recipients, secondary: Optional[Recipients] = None)

Bases: Feature

Feature that controls where a HITL interrupt is delivered.

Routing specifies a primary group of recipients (required) and an optional secondary group used for escalation when a Ttl feature with on_expiry=OnExpiry.FORWARD_TO_SECONDARY is also active.

Usage example::

import hgateway_sdk as hg

features = hg.RunFeatures(
    routing=hg.Routing(
        primary=hg.Recipients(channels=["#approvals"]),
        secondary=hg.Recipients(users=["manager@example.com"]),
    )
)
hg.raise_interrupt("deploy_approval", content, features=features)

Attributes: primary: The first-line Recipients group to notify. Must contain at least one channel or user; validation raises an error otherwise. secondary: Optional fallback Recipients group used when Ttl.on_expiry=OnExpiry.FORWARD_TO_SECONDARY. Defaults to None.

hgateway_sdk.Recipients dataclass

Recipients(channels: list[str] = list(), users: list[str] = list())

A group of notification targets (channels and/or named users).

Used as the primary and optional secondary targets inside a Routing feature. At least one of channels or users must be non-empty for a Recipients instance to be considered valid by Routing.validate().

Usage example (nested inside Routing)::

import hgateway_sdk as hg

routing = hg.Routing(
    primary=hg.Recipients(channels=["#ops-approvals"], users=["alice@example.com"]),
    secondary=hg.Recipients(channels=["#escalations"]),
)
features = hg.RunFeatures(routing=routing)

Attributes: channels: List of notification channel identifiers (e.g. Slack channel names or IDs) to which the interrupt is delivered. Defaults to an empty list. users: List of individual user identifiers (e.g. email addresses or user IDs) to whom the interrupt is delivered. Defaults to an empty list.

hgateway_sdk.Ttl dataclass

Ttl(seconds: int, on_expiry: OnExpiry = OnExpiry.DEFAULT_RESPONSE, default_response: Optional[dict] = None)

Bases: Feature

Feature that enforces a response deadline on a HITL interrupt.

When the operator does not respond within seconds, the gateway acts according to on_expiry:

• OnExpiry.DEFAULT_RESPONSE — auto-resolve with default_response.
• OnExpiry.FORWARD_TO_SECONDARY — escalate to routing.secondary (requires a Routing feature with a non-empty secondary).

Usage example::

import hgateway_sdk as hg
from hgateway_sdk.constants import OnExpiry

features = hg.RunFeatures(
    routing=hg.Routing(
        primary=hg.Recipients(channels=["#approvals"]),
        secondary=hg.Recipients(channels=["#escalations"]),
    ),
    ttl=hg.Ttl(
        seconds=3600,
        on_expiry=OnExpiry.FORWARD_TO_SECONDARY,
    ),
)

Attributes: seconds: Number of seconds the gateway waits for an operator response before applying the on_expiry action. on_expiry: OnExpiry enum value specifying what happens when the deadline passes. Defaults to OnExpiry.DEFAULT_RESPONSE. default_response: Pre-built response dict used when on_expiry=OnExpiry.DEFAULT_RESPONSE. Must be provided when that expiry mode is active; validation raises an error otherwise. Defaults to None.

hgateway_sdk.Feature

Bases: ABC

A run-overridable feature. Owns its own (intra-feature) validation, and may emit cross-feature rules that constrain other features.

cross_rules

cross_rules() -> list[CrossFeatureRule]

Constraints this feature imposes on other features (default: none).

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/schemas/features/feature.py

def cross_rules(self) -> list[CrossFeatureRule]:
    """Constraints this feature imposes on other features (default: none)."""
    return []

validate abstractmethod

validate() -> list[str]

Errors internal to this feature alone.

Source code in .venv/lib/python3.12/site-packages/hgateway_sdk/schemas/features/feature.py

@abstractmethod
def validate(self) -> list[str]:
    """Errors internal to this feature alone."""

hgateway_sdk.CrossFeatureRule

Bases: ABC

A constraint one feature imposes on another. Owned by (returned from cross_rules() of) the trigger feature, and checked against the full feature set.

hgateway_sdk.HitlSpec dataclass

HitlSpec(hitl_key: str, content: HitlContent, runtime: RuntimeContext, features: RunFeatures = RunFeatures(), spec_version: str = constants.SPEC_VERSION)

Complete wire payload sent to the HITL gateway for a single interrupt.

The SDK serialises this dataclass to JSON and POSTs it to the gateway's register endpoint. All fields are assembled automatically by the client from the arguments passed to raise_interrupt().

Attributes: hitl_key: Logical, stable name of this interrupt point in the agent graph (e.g. "send_email_approval"). Used by the gateway for routing and deduplication.

    **Treat this as immutable once an interrupt point is in use.** It is
    part of the interrupt's dedup identity
    (``hitl_key`` + ``checkpoint_ns`` + ``occurrence``), so the gateway
    relies on it staying constant across a graph's pause/resume cycles
    and across redeploys. The same logical interrupt **must** pass the
    same ``hitl_key`` every time it is raised — on resume the node is
    replayed and ``raise_interrupt`` re-registers, and a changed key
    reads as a *different* interrupt (breaking dedup, correlation, and
    any in-flight gateway-owned interaction). Do not derive it from
    anything that varies per run (timestamps, UUIDs, counters, state
    values); use a hard-coded literal per interrupt site.
content: The content object describing what the operator should do —
    one of the 10 concrete ``HitlContent`` subclasses.
runtime: ``RuntimeContext`` carrying the LangGraph thread/run
    identifiers and current graph state snapshot.
features: Optional ``RunFeatures`` controlling routing, TTL, and other
    per-interrupt behaviour.  Defaults to an empty ``RunFeatures()``.
spec_version: Protocol version string embedded in every payload so the
    gateway can handle schema evolution.  Defaults to the SDK constant
    ``SPEC_VERSION`` (currently ``"1.1"``).