Metadata-Version: 2.4
Name: hgateway-sdk
Version: 0.1.1
Summary: theved.ai HITL Gateway SDK — raise human-in-the-loop interrupts from LangGraph nodes.
Project-URL: Homepage, https://theved.ai
Project-URL: Source, https://github.com/theved-ai/hgateway_sdk
Author: theved.ai
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: langchain-core>=1.0
Requires-Dist: langgraph>=1.0
Requires-Dist: python-dotenv>=1.2.2
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == 'dev'
Description-Content-Type: text/markdown

# 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:

```bash
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](https://github.com/theved-ai/hgateway_sdk/releases); 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

```python
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:

```python
{"__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 `TypedDict`s — access them by key (`resp["decision"]`), never by
attribute.

## Recipes

### A decision

```python
import hgateway_sdk as hg
from hgateway_sdk import SingleDecisionContent, Option

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

### An approval with routing + TTL

```python
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=["manager@example.com"]),
    ),
    ttl=Ttl(seconds=1800, on_expiry=OnExpiry.FORWARD_TO_SECONDARY),
)
resp = hg.raise_interrupt("deploy-approval", BinaryApprovalContent(prompt="Deploy?"), features=features)
```

### The fallback pattern

```python
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:

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