Skip to content

Integration Guide

Three points of integration: setting things up on the dashboard, wiring your agent, and wiring your BFA (the backend that receives the resume). See Architecture & Lifecycle for how these connect end-to-end.

Pre-requisite: dashboard setup

Before any code, set up your tenant and agent on dashboard:

  1. Sign in — Google SSO only, no password to manage.
  2. Create your organization — the tenant boundary everything else (agents, keys, channels, credentials) lives under.
  3. Create an agent — one entry per agent you're integrating with Ved.

  4. Generate an API key — shown once as HGATEWAY_AGENT_API_KEY=...; this is what the SDK authenticates with.

    One key per agent, not one key for your whole tenant

    HGATEWAY_AGENT_API_KEY is scoped to the single agent you just created, not a tenant-wide credential. Ved only ever talks to specific agents, so every agent you integrate gets its own key — generate a separate one per agent, don't reuse a single key across multiple agents.

  5. Set the resume callback — the URL on your BFA that Ved will POST a resume to, plus a signing secret (either dashboard-minted or your own) used to HMAC-sign that callback. Copy the shown secret as HGATEWAY_CALLBACK_SIGNING_SECRET — see Auth: resume for the exact signature scheme your BFA needs to verify.

    Two secrets to store carefully, not one

    By this point you're holding two separate credentials for this agent — HGATEWAY_AGENT_API_KEY (step 4 above) and HGATEWAY_CALLBACK_SIGNING_SECRET (this step). Both are shown once and both are sensitive: the API key authenticates your agent to Ved, the signing secret is what lets your BFA verify a resume callback actually came from Ved. Store both securely on your side — the dashboard won't show either again after generation.

  6. Install the SDK in your Agent:

    pip install hgateway-sdk

  7. Configure the key: add HGATEWAY_AGENT_API_KEY as an environment variable in your project (or a .env file — it's read either way). That's the only required setting; a missing key raises GatewayBootstrapError. Optionally, set HGATEWAY_TIMEOUT_SECONDS to change the request timeout (default 5).

Agent-level integration

Two steps, both on the agent side that calls the SDK:

1. Decorate your node

import hgateway_sdk as hg

@hg.hitl_node(channels=["#ops"], ttl_seconds=3600, on_expiry="default-response")
def review_node(state):
    ...

There's no setup call before this — raise_interrupt auto-initializes the gateway client from HGATEWAY_AGENT_API_KEY on first use. You only need hg.init(transport=...) if you're injecting a custom transport (e.g. a fake for tests) or want agent-wide routing/ttl defaults; see Client & Decorators.

@hitl_node also takes the same routing/ttl kwargs as init() and raise_interrupt — every HITL raised from this node inherits them unless overridden at the call site. Routing and Ttl are mandatory: some scope (init, hitl_node, or the call site) must supply both before a HITL can be raised. If this node needs no defaults of its own, @hg.hitl_node (no parens) still works, same as before.

@hitl_node binds the current LangGraph state, config, and an occurrence counter into contextvars, so the SDK can build the RuntimeContext automatically when you call raise_interrupt inside it. Calling raise_interrupt outside a @hitl_node-decorated node raises HitlNodeRequired — this decorator always comes first.

It works the same way on a plain function or on an instance method (e.g. a class's __call__):

class ReviewNode:
    @hg.hitl_node(channels=["#ops"], ttl_seconds=3600, on_expiry="default-response")
    def __call__(self, state):
        resp = hg.raise_approval("review", "Approve this action?")
        return {"result": resp}

Only decorate the method actually registered as the node — never the class itself. There's no reliable way to tell which methods on a class are LangGraph nodes versus plain helpers, so class-level decoration isn't supported.

2. Raise the interrupt

Every interrupt is one of 4 families (HitlType): Approval, Decision, Context, Edit. For each concrete content type there's a matching raise_* wrapper function that builds the content object and calls raise_interrupt for you — you don't need to import or construct a content class yourself. One worked example per family (see Wrapper Functions for all 10):

Approval — raise_approval

resp = hg.raise_approval(
    "approve-action",
    f"Approve this action? {state['summary']}",
    fallback={"decision": "reject"},
)
# response: BinaryApprovalResponse -> {"decision": str}

Decision — raise_pick_one_option

from hgateway_sdk import Choice

resp = hg.raise_pick_one_option(
    "pick-region",
    "Which region should we deploy to?",
    options=[Choice("us", "US"), Choice("eu", "EU")],
)
# response: SingleDecisionResponse -> {"selected": str}

Context — raise_req_for_context

resp = hg.raise_req_for_context(
    "customer-preference",
    "What is the customer's preferred resolution?",
)
# response: FreetextContextResponse -> {"value": str}

Edit — raise_draft

resp = hg.raise_draft(
    "review-email",
    "Review the generated email and edit if needed before sending.",
    draft="Hi Jane,\n\nWe're following up on your recent support request…\n\nBest, Support Team",
)
# response: ContentEditResponse -> {"content": str}

If you need a content type with no wrapper, or want to build the content object yourself, call hg.raise_interrupt(hitl_key, content, ...) directly — see Content Schemas for the full class table.

  • hitl_key (first positional arg) is the unique identifier of this interrupt call site, per agent — every HITL raised from this exact spot in your code, across every run, is grouped under this one key. It should stay the same across resumes of the same node — it's part of the gateway's dedup identity.
  • Routing and ttl are mandatory. Every raise_* function (and raise_interrupt) accepts channels, users, secondary_channels, secondary_users, ttl_seconds, on_expiry, default_response, or a full features=RunFeatures(...). These resolve through the same cascade as @hitl_node (init < hitl_node < call site) — set them here only if this specific call needs to override what the node or init() already provided. If no scope ever supplies both routing and ttl, FeatureValidationError is raised. See Features & Routing for the full cascade and the CLEAR_FEATURE sentinel.
  • fallback is an optional reliability parameter. Even if Ved is unreachable, your agent's HITLs should never just stop — fallback is how you guarantee that. If the gateway is unreachable or declines ownership, the call raises the interrupt locally with the fallback content instead, which your BFA can handle itself. Ved owning the interaction (routing, notification, TTL) is the common case; fallback is what keeps the agent working without a hard dependency on Ved when it isn't.

BFA-level integration

1. Verify the resume callback's signature

Add HGATEWAY_CALLBACK_SIGNING_SECRET (the secret from the dashboard Pre-requisite step above) as an env var in your BFA, and use it to verify every resume request before trusting it — see Auth: resume for the full header/signature spec this implements:

import hmac
import hashlib
import os
import time

SIGNING_SECRET = os.environ["HGATEWAY_CALLBACK_SIGNING_SECRET"]

def verify_payload(raw_body: bytes, timestamp: str, signature: str) -> bool:
    if abs(time.time() - int(timestamp)) > 60 * 5:
        return False  # stale request — replay protection
    basestring = f"v0:{timestamp}:{raw_body.decode()}"
    computed = "v0=" + hmac.new(
        SIGNING_SECRET.encode(), basestring.encode(), hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed, signature)

Call verify_payload(raw_body, request.headers["X-Ved-Request-Timestamp"], request.headers["X-Ved-Signature"]) against the raw request body bytes (not a re-parsed/re-serialized copy) before doing anything with a resume payload.

2. Don't double-handle a gateway-owned interrupt

Your BFA (see Architecture & Lifecycle) will see every suspended graph, including ones the gateway has taken over. Check the interrupt value before rendering your own UI or resolving it yourself:

state = graph.get_state(config)
for task in state.tasks:
    for i in task.interrupts:
        if i.value.get("__hgateway_owned__"):
            # The gateway is driving this one and collecting the response. Don't render anything; 
            # just wait for the gateway's signed resume callback (see Architecture & Lifecycle).
            continue
        # Not gateway-owned (declined/unreachable fallback, or a plain local
        # interrupt) — handle it yourself as you normally would.

If you skip this check, a fallback-unaware BFA will try to resolve an interrupt the gateway already owns, racing the gateway's own resume.

Common pitfalls

  • HitlNodeRequired — raise_interrupt called outside an @hitl_node node. Decorate the calling node.
  • GatewayBootstrapError — HGATEWAY_AGENT_API_KEY isn't set. Set the env var or provide a .env.
  • FeatureValidationError — no scope in the cascade (init, hitl_node, call site) supplied both routing and ttl. Set them at init() for an agent-wide default, or per node/call if they vary.
  • TtlForwardNeedsSecondary — you set Ttl.on_expiry=OnExpiry.FORWARD_TO_SECONDARY without also setting Routing.secondary — the gateway has nowhere to escalate to.

See Client & Decorators and Constants & Exceptions for the full API surface.