Metadata-Version: 2.4
Name: langgraph-kirokuforms
Version: 0.3.0
Summary: Human-in-the-Loop integration between LangGraph and KirokuForms
Author-email: KirokuForms Team <info@kirokuforms.com>
License: MIT
Project-URL: Homepage, https://www.kirokuforms.com/ai/langgraph
Project-URL: Documentation, https://chelseaaiventures.github.io/langgraph-kirokuforms/
Project-URL: Repository, https://github.com/ChelseaAIVentures/langgraph-kirokuforms
Project-URL: Issues, https://github.com/ChelseaAIVentures/langgraph-kirokuforms/issues
Keywords: langgraph,human-in-the-loop,hitl,forms,mcp,approval
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
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
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: requests>=2.28.0
Requires-Dist: langgraph>=1.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.18.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: isort>=5.0.0; extra == "dev"
Requires-Dist: mypy>=0.900; extra == "dev"
Requires-Dist: build>=1.0.0; extra == "dev"
Requires-Dist: twine>=5.0.0; extra == "dev"
Requires-Dist: tomli>=2.0; python_version < "3.11" and extra == "dev"

# LangGraph KirokuForms Integration

**Human-in-the-Loop integration between LangGraph and KirokuForms**

This library lets a LangGraph workflow pause, ask a human to fill in a form, and resume with the answers. It talks to KirokuForms over the Model Context Protocol (MCP) endpoints at `https://www.kirokuforms.com/api/mcp`.

## Installation

Not on PyPI yet. Install from the repository:

```bash
pip install git+https://github.com/ChelseaAIVentures/langgraph-kirokuforms.git
```

Once `0.2.0` is published (see [RELEASING.md](RELEASING.md)) this becomes
`pip install langgraph-kirokuforms`, and this paragraph goes away.

The distribution is named `langgraph-kirokuforms`, but the import path is `kirokuforms`:

```python
from kirokuforms import KirokuFormsHITL, create_kiroku_interrupt_node
```

Needs Python 3.10 or newer and LangGraph 1.0 or newer. Tested against LangGraph
1.1.6. The REST client itself has no LangGraph dependency at import time, so you
can use `KirokuFormsHITL` on its own from anything.

### API key scopes

Create a key in the KirokuForms dashboard. The endpoints this client calls check scopes:

| Client method                          | Scopes required               |
| -------------------------------------- | ----------------------------- |
| `create_task`, `create_verification_task` | `hitl:create` and `forms:write` |
| `get_task_result`, `list_tasks`        | `hitl:read`                   |
| `cancel_task`                          | `hitl:create`                 |

A key missing a scope gets a 401 naming the scope it needs. Creating and reading tasks works on every plan, including the free one. The one part that is paid is assigning a task to an email address with no KirokuForms account, which a free-plan key is refused with a 402 (see [Assigning a task to a person](#assigning-a-task-to-a-person)).

## Quick Start

```python
from kirokuforms import KirokuFormsHITL, create_kiroku_interrupt_node
from langgraph.graph import StateGraph

# base_url defaults to https://www.kirokuforms.com/api/mcp, so it can be omitted.
client = KirokuFormsHITL(api_key="your-api-key")

request_verification = create_kiroku_interrupt_node(
    client,
    name="verification",
    title="Verify Data",
    description="Please verify this information is correct",
    fields=[
        {
            "type": "text",
            "label": "Customer Name",
            "name": "customer_name",
            "required": True,
        },
        {
            "type": "radio",
            "label": "Information is Correct",
            "name": "is_correct",
            "required": True,
            "options": [
                {"label": "Yes", "value": "yes"},
                {"label": "No", "value": "no"},
            ],
        },
    ],
)

workflow = StateGraph(dict)
workflow.add_node("request_verification", request_verification)
```

The node calls LangGraph's `interrupt()`, so invoking the graph stops at this
step and hands back the link to send the reviewer. Compile with a checkpointer
and invoke with a `thread_id`, or there is nothing to suspend into. See
[Suspending the graph](#suspending-the-graph) for the resume half.

The handler merges its outcome into the state under `human_verification`:

```python
{
    "human_verification": {
        "completed": True,               # False if it timed out, or wait_for_result was False
        "task_id": "task_1752...",       # the external task ID
        "form_url": "https://www.kirokuforms.com/hitl/task/...?token=...",
        "result": {"customer_name": "Acme", "is_correct": "yes"},  # None until completed
    }
}
```

`form_url` carries a single-use access token. It is the link you send to the reviewer, and it is the only copy of that token: KirokuForms stores only a hash of it, so it cannot be read back out of the API later.

## Core Concepts

### Model Context Protocol (MCP)

KirokuForms implements the Model Context Protocol, a standard interface for AI systems to request human input. A LangGraph workflow uses it to:

1. Request human review of specific data
2. Generate a form for the human reviewer
3. Receive and process the human's answers
4. Resume execution with that feedback

### Two ways to wait for a human

There are two, and the difference is what happens to your process while the
human is asleep.

**`create_kiroku_interrupt_node` suspends the graph.** It uses LangGraph's own
`interrupt()`, so the run stops, the state is written to your checkpointer, and
nothing is held open. Resume it minutes or days later, from a different process.
This is the one to use.

**`create_kiroku_interrupt_handler` blocks the node. Deprecated since 0.3.0.**
It creates the task and then polls in a loop, by default for up to an hour, with
a worker held open the whole time. It is not an interrupt in LangGraph's sense:
the graph is never suspended, nothing is checkpointed, and nothing can resume in
another process. It still works and still receives fixes, because published code
imports it, but it will not gain features and it emits a `DeprecationWarning`.
Migration is usually four lines, and both write to `state["human_verification"]`
so nothing downstream changes.

### Suspending the graph

```python
from kirokuforms import KirokuFormsHITL, create_kiroku_interrupt_node, resume_with_answers
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

client = KirokuFormsHITL(api_key="your-api-key")

review = create_kiroku_interrupt_node(
    client,
    name="approval",                      # distinguishes this step's tasks
    title="Approve the invoice",
    description="Check the amount before we pay it.",
    fields=[
        {
            "type": "radio",
            "label": "Approve?",
            "name": "approved",
            "required": True,
            "options": [
                {"label": "Approve", "value": "yes"},
                {"label": "Reject", "value": "no"},
            ],
        },
    ],
    assign_to_email="jane@acme.com",      # optional: send it straight to Jane
)

builder = StateGraph(dict)
builder.add_node("review", review)
builder.add_edge(START, "review")
builder.add_edge("review", END)

# A checkpointer is required. Without one there is nothing to suspend into.
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "invoice-42"}}

state = graph.invoke({"amount": 1200}, config)
# The run has stopped. The payload tells you where to send the human.
payload = state["__interrupt__"][0].value
print(payload["form_url"])       # tokenized link, send this to the reviewer
print(payload["kiroku_task_id"])
```

Later, when the human has answered (your webhook fired, or you polled):

```python
final = graph.invoke(resume_with_answers(client, payload["kiroku_task_id"]), config)
print(final["human_verification"]["result"])   # {"approved": "yes"}
```

`resume_with_answers` reads the submitted answers back from the task and wraps
them in a `Command`. If you would rather supply them yourself, plain
`Command(resume={...})` works the same way.

#### The re-execution rule

When LangGraph resumes a graph, **the interrupted node runs again from its first
line**. `interrupt()` returns the resume value this time instead of suspending,
but everything above it happens a second time.

That would ordinarily mean a second task, a second form and a second email to
your reviewer. It does not here: the node derives a task id from the graph's
thread and sends it as an `Idempotency-Key`, and a repeat create replays the
case that already exists. You can watch this happen in
`tests/test_graph.py::test_resuming_does_not_create_a_second_case`, which
asserts two calls and one case.

The consequence is worth knowing: one node interrupting twice in the same
thread, in a loop, would reuse a single case. Pass `task_id_for=lambda state:
...` to make the id depend on whatever separates the iterations.

### Interrupt handlers (blocking, deprecated)

> Deprecated since 0.3.0. Use [`create_kiroku_interrupt_node`](#suspending-the-graph). This section documents what existing code does, not what new code should do.

`create_kiroku_interrupt_handler` returns a function you call inside a node. It creates the task, and, if `wait_for_result` is true, polls until the human submits, then writes the answers into the state.

The interrupt dictionary you pass it accepts:

| Key               | Default                        | Meaning                                                    |
| ----------------- | ------------------------------ | ---------------------------------------------------------- |
| `title`           | "Human Verification Required"   | Task title, shown to the reviewer                          |
| `description`     | "Please verify the following information" | Instructions for the reviewer                     |
| `fields`          | `[]`                           | Field definitions (see Field Types)                        |
| `data`            | `{}`                           | Used only when `fields` is empty: fields are generated from it |
| `wait_for_result` | `True`                         | Block until the human submits, or return immediately       |

With `wait_for_result: False` the node returns as soon as the task exists, with `completed: False` and `result: None`. Collect the answers later through the webhook (`callback_url`) or by polling `get_task_result`.

## API Reference

### `KirokuFormsHITL`

```python
KirokuFormsHITL(
    api_key: str,
    base_url: str = "https://www.kirokuforms.com/api/mcp",
    webhook_url: Optional[str] = None,
    webhook_secret: Optional[str] = None,
    timeout: int = 10,
    max_retries: int = 3,
)
```

- **api_key**: Your KirokuForms API key
- **base_url**: The MCP API base URL. Point it at `http://127.0.0.1:4321/api/mcp` to run against a local dev server. Use the literal address, not `localhost`: that name resolves to both 127.0.0.1 and `[::1]`, and if anything else on the machine is listening on the other one, some of your requests quietly reach it instead.
- **webhook_url**: Default `callback_url` for tasks created by this client
- **webhook_secret**: Secret for webhook verification
- **timeout**: Request timeout in seconds
- **max_retries**: Maximum retries for a failed request

#### `create_task`

```python
create_task(
    title: str,
    description: str = "",
    fields: Optional[List[Dict[str, Any]]] = None,
    template_id: Optional[str] = None,
    initial_data: Optional[Dict[str, Any]] = None,
    expiration: Optional[str] = None,
    priority: str = "medium",
    task_id: Optional[str] = None,
    callback_url: Optional[str] = None,
    assign_to_email: Optional[str] = None,
    assignee_name: Optional[str] = None,
    assign_to_slack: Optional[bool] = None,
    idempotency_key: Optional[str] = None,
) -> Dict[str, Any]
```

Creates a human-in-the-loop task. Either `fields` or `template_id` is required.

- **title**: The title of the task
- **description**: Instructions for the human reviewer
- **fields**: Field definitions (see Field Types below)
- **template_id**: An existing form to use instead of `fields`
- **initial_data**: Pre-filled values for the form
- **expiration**: How long until the task expires, as digits plus `h` or `d` (`"24h"`, `"3d"`)
- **priority**: `"low"`, `"medium"`, or `"high"`
- **task_id**: Your own ID for the task, instead of a generated one
- **callback_url**: URL to POST to when the task completes
- **assign_to_email**: Hand the task to this address (see [Assigning a task to a person](#assigning-a-task-to-a-person))
- **assignee_name**: Display name for the assignee, shown in the notification
- **assign_to_slack**: Channel choice for the assignee: unset or `True` DMs them on Slack when the address is in the owner's workspace and emails them otherwise, `False` emails a workspace teammate instead of DMing them
- **idempotency_key**: Sent as the `Idempotency-Key` header. Repeating a create with the same key replays the case the first call made, with the same `formUrl` and the same access token, rather than minting a second form, task, token and email. Use it whenever a create can run twice for reasons outside your control: a retried request, a redelivered queue message, or a LangGraph node re-executing after a resume. `task_id` also works as one, and the server honours either, but the header does not double as the task's name

Returns:

- **taskId**: The external task ID, used by every other method here
- **hitlTaskId**: The internal task ID
- **formId**: The ID of the review page built for this task. With `fields`, it is
  created per task and does **not** count against the account's form allowance,
  and it does not show up in the owner's forms list. With `template_id`, it is
  the id you passed, an ordinary form the owner keeps and which does count.
- **formUrl**: The tokenized URL where the human completes the task
- **expiresAt**: When the task expires, if `expiration` was set

##### What a task costs against the plan

A completed review is a submission, and counts against the account's monthly
submission allowance on the [pricing page](https://www.kirokuforms.com/pricing).
That is the whole of it. Creating a task costs nothing until somebody answers it.

Two safety rails exist to stop a looping agent, not to separate the plans:
reviews created per day, and reviews left open at once. They are set far above
any real queue, so you should not meet one. If you do, the client raises a
`ValueError` naming `RATE_LIMIT_EXCEEDED` or `CONCURRENT_TASK_LIMIT_EXCEEDED`
from a `429`. It retries a 429 with backoff, which suits the daily rail; for the
concurrency one, retrying will not help until a case is completed or canceled,
and hitting it at all is a sign something is creating cases in a loop.

##### Assigning a task to a person

`assign_to_email` hands a task to a specific person by address. Two things determine what happens next: whether that address already has a KirokuForms account, and whether Slack is connected.

- The address belongs to an existing account: the task is assigned to that teammate and shown to them in-app. This is free.
- The address has no account: the recipient gets a tokenized link and completes the task without signing up. This is the paid part. A key on a free plan is refused with a 402 (`SUBSCRIPTION_REQUIRED`); assigning a teammate who has an account is not gated.

`assign_to_slack` picks the delivery channel. Left unset (or `True`), a resolvable Slack user in the owner's workspace is DMed the task with Approve / Reject / Open buttons, and anyone else is emailed. `False` emails a workspace teammate instead of DMing them. The tokenized link is only ever sent in a DM or an email, never posted to a shared channel, so it stays a per-recipient credential.

```python
client.create_task(
    title="Approve the Q3 invoice",
    description="Please review and approve this invoice.",
    assign_to_email="jane@acme.com",
    assignee_name="Jane Doe",
    fields=[
        {
            "type": "radio",
            "label": "Approve?",
            "name": "approved",
            "required": True,
            "options": [
                {"label": "Approve", "value": "approve"},
                {"label": "Reject", "value": "reject"},
            ],
        },
    ],
)
```

The endpoint reference for these settings is at [kirokuforms.com/ai/mcp/tools/hitl](https://www.kirokuforms.com/ai/mcp/tools/hitl).

#### `create_verification_task`

```python
create_verification_task(
    title: str,
    description: str,
    data: Dict[str, Any],
    fields: Optional[List[Dict[str, Any]]] = None,
    **kwargs: Any,
) -> Dict[str, Any]
```

Wraps `create_task`. When `fields` is omitted, it builds one field per key in `data` (bools become radios, numbers become number inputs, everything else becomes text), then appends an "Is this information correct?" radio and a comments textarea. Extra keyword arguments pass through to `create_task`.

#### `get_task_result`

```python
get_task_result(
    task_id: str,
    wait: bool = True,
    timeout: int = 3600,
) -> Dict[str, Any]
```

Returns the submitted form data as a flat dictionary of field name to value, once the task is completed. With `wait=True` it polls every 5 seconds until then.

A task with no submission comes back as a status report instead:

```python
{"taskId": "task_abc", "status": "pending", "completed": False, "data": None}
```

You get that report in two cases: `wait=False` and the human has not answered yet, or the task reached a status it can never come back from (`canceled`, `expired`), where polling would only burn the timeout. Check `result.get("completed") is False` to tell a report from a submission.

`TimeoutError` is raised only when `wait=True` and `timeout` elapses with the task still pending.

```python
result = client.get_task_result(task_id, wait=False)
if result.get("completed") is False:
    print(f"still {result['status']}")
else:
    print(f"the human said: {result}")
```

#### `list_tasks`

```python
list_tasks(
    status: Optional[str] = None,
    limit: int = 10,
    offset: int = 0,
) -> Dict[str, Any]
```

Lists your own tasks, newest first.

- **status**: Filter by `"pending"`, `"completed"`, `"canceled"`, or `"expired"`
- **limit**: Page size, 1 to 100
- **offset**: Pagination offset

Returns `{"tasks": [...], "total": int, "limit": int, "offset": int}`. Each task carries `id`, `taskId`, `title`, `description`, `status`, `priority`, `createdAt`, `expiresAt`, `completedAt`, `callbackUrl`, and `metadata`. Access tokens are never included.

#### `cancel_task`

```python
cancel_task(task_id: str) -> Dict[str, Any]
```

Cancels a pending task and revokes its access tokens, so a link already emailed to a reviewer stops working. Returns `{"id", "taskId", "title", "status"}` with `status` set to `"canceled"`.

Only a pending task can be canceled. A task that is already completed, canceled, or expired returns a 409 (`TASK_NOT_PENDING`) and is left untouched, so cancelling cannot discard a submission a human already made.

### `create_kiroku_interrupt_handler` (deprecated)

```python
create_kiroku_interrupt_handler(
    api_key: str,
    **kwargs: Any,
) -> Callable[[Dict[str, Any], Dict[str, Any]], Dict[str, Any]]
```

Builds a `KirokuFormsHITL` from `api_key` plus any constructor keyword arguments, and returns an interrupt handler you call as `handler(state, interrupt_data)`.

### Verifying webhooks

A `callback_url` receives an unauthenticated POST from the internet. Anything
that learns the URL can post to it, so a handler that trusts the body will
happily record an approval nobody gave.

Set a secret on the webhook in KirokuForms, pass the same string as
`webhook_secret`, and check every delivery:

```python
from kirokuforms import KirokuFormsHITL, WebhookVerificationError

client = KirokuFormsHITL(api_key="…", webhook_secret="whsec_…")

@app.post("/kiroku-webhook")
def kiroku_webhook(request):
    try:
        payload = client.verify_webhook(
            request.body,                                    # raw bytes or str
            request.headers.get("X-KirokuForms-Signature-256"),
        )
    except WebhookVerificationError as exc:
        return Response(status=400, body=str(exc))

    if payload["eventType"] == "hitl.task.completed":
        answers = payload["data"]["submission"]["data"]
        ...
    return Response(status=200)
```

KirokuForms signs with `HMAC-SHA256(secret, body-without-its-signature-field)`,
hex encoded, and sends the digest both in the `X-KirokuForms-Signature-256`
header and as a `signature` field inside the JSON. Passing the header is
preferred; omit it and the field is used instead, which is what you want in a
framework that hands you parsed JSON and no headers.

`verify_webhook` is also importable on its own if you would rather not build a
client just to check a signature:

```python
from kirokuforms import verify_webhook
payload = verify_webhook(raw_body, "whsec_…", signature_header)
```

Comparison is constant-time. A wrong secret, a missing signature, a body that is
not JSON, or a body edited after signing all raise `WebhookVerificationError`.

> Before 0.2.0, `webhook_secret` was accepted, documented as being for
> verification, and read by nothing. If you built a callback endpoint from an
> earlier README, it is not verifying anything; add the call above.

### Errors

- **`ValueError`**: the API refused the request. The message carries the API's own error code and the HTTP status, e.g. `API Error (TASK_NOT_PENDING): this task is canceled [HTTP 409]`. A refusal is an answer, so it is raised on the first response; it is not retried.
- **`ConnectionError`**: the request never got an answer (DNS, refused connection, read timeout), or a 5xx / 429 was still failing after `max_retries`. These are the only failures a retry can fix, so they are the only ones retried, with exponential backoff.
- **`TimeoutError`**: `get_task_result` gave up waiting for the human.

A task ID belonging to another account returns 404, the same answer as a task ID that does not exist, so a guessed ID reveals nothing.

```python
try:
    client.cancel_task(task_id)
except ValueError as refusal:
    if "TASK_NOT_PENDING" in str(refusal):
        print("someone already answered or canceled it")
```

## Field Types

`type` defaults to `text`. Accepted values: `text`, `email`, `number`, `tel`, `url`, `password`, `textarea`, `select`, `checkbox`, `radio`, `date`, `time`, `file`, `hidden`, `honeypot`.

Each field accepts these properties:

| Property       | Required | Notes                                                            |
| -------------- | -------- | ---------------------------------------------------------------- |
| `label`        | yes      | Shown above the input                                            |
| `name`         | yes      | The key this field's answer arrives under                        |
| `type`         | no       | One of the types above; defaults to `text`                       |
| `required`     | no       | Defaults to `False`                                              |
| `defaultValue` | no       | **A string**, for every type. Send `str(1245.0)`, not `1245.0`.  |
| `placeholder`  | no       | Placeholder text                                                 |
| `options`      | for `select`, `radio`, `checkbox` | List of `{"label": ..., "value": ...}` |
| `validation`   | no       | Validation rules object                                          |
| `helpText`     | no       | Hint shown under the input                                       |
| `className`    | no       | Extra CSS classes                                                |
| `order`        | no       | Display order; defaults to the order of the list                 |

A numeric or boolean `defaultValue` is rejected with a 400 (`VALIDATION_ERROR`). Unknown properties are dropped silently.

## Examples

### Create a task and collect the result

```python
from kirokuforms import KirokuFormsHITL

client = KirokuFormsHITL(api_key="your-api-key")

response = client.create_task(
    title="Verify Transaction",
    description="Please review this transaction for approval",
    expiration="24h",
    fields=[
        {
            "type": "text",
            "label": "Transaction ID",
            "name": "transaction_id",
            "required": True,
            "defaultValue": "TRX-12345",
        },
        {
            "type": "number",
            "label": "Amount",
            "name": "amount",
            "required": True,
            "defaultValue": "1245.00",
        },
        {
            "type": "radio",
            "label": "Approve Transaction?",
            "name": "approved",
            "required": True,
            "options": [
                {"label": "Approve", "value": "approve"},
                {"label": "Reject", "value": "reject"},
            ],
        },
    ],
)

# Send this link to the reviewer. It carries the access token.
print(f"Review this transaction at: {response['formUrl']}")

task_id = response["taskId"]

# Block until a human submits, for up to an hour.
result = client.get_task_result(task_id, timeout=3600)
print(f"Approved: {result['approved']}")
```

To check on it without blocking, list it instead of polling for the result:

```python
task = client.list_tasks(limit=1)["tasks"][0]
print(f"Task status: {task['status']}")

if task["status"] == "pending":
    client.cancel_task(task["taskId"])  # revokes the reviewer's link
```

### Asynchronous workflow with a webhook

```python
from kirokuforms import KirokuFormsHITL, create_kiroku_interrupt_node
from langgraph.graph import StateGraph

client = KirokuFormsHITL(
    api_key="your-api-key",
    webhook_url="https://your-server.com/webhook/langgraph",
    webhook_secret="your-webhook-secret",
)

def process_data(state):
    return {"processed_data": state["input_data"]}

# The graph suspends here. Your process is free; the webhook fires when the
# human answers, and that handler resumes the thread.
#
# `fields` is fixed when the node is built, so it cannot interpolate state.
# To show the reviewer values from this run, pass `data_key="processed_data"`
# instead and the fields are generated from that dict per case.
request_human_review = create_kiroku_interrupt_node(
    client,
    name="content-review",
    title="Review Generated Content",
    description="Please review the following content for accuracy",
    fields=[
        {
            "type": "textarea",
            "label": "Generated Content",
            "name": "content",
            "required": True,
        },
        {
            "type": "radio",
            "label": "Content Quality",
            "name": "quality",
            "required": True,
            "options": [
                {"label": "Excellent", "value": "excellent"},
                {"label": "Good", "value": "good"},
                {"label": "Needs Improvement", "value": "needs_improvement"},
            ],
        },
    ],
)

def handle_approved(state):
    return {"status": "approved", "final_data": state["human_verification"]["result"]}

def handle_rejected(state):
    return {"status": "rejected", "reason": state["human_verification"]["result"]["quality"]}

workflow = StateGraph(dict)
workflow.add_node("process_data", process_data)
workflow.add_node("request_review", request_human_review)
workflow.add_node("handle_approved", handle_approved)
workflow.add_node("handle_rejected", handle_rejected)

workflow.add_edge("process_data", "request_review")
workflow.add_conditional_edges(
    "request_review",
    lambda state: (
        "approved"
        if state["human_verification"]["result"]["quality"] in ("excellent", "good")
        else "rejected"
    ),
    {"approved": "handle_approved", "rejected": "handle_rejected"},
)

app = workflow.compile()
```

A conditional edge that reads `human_verification["result"]` needs the answers to be there, so it belongs downstream of a `wait_for_result: True` review, or of the webhook that resumes the graph.

## Development

1. Clone the repository

   ```bash
   git clone https://github.com/ChelseaAIVentures/langgraph-kirokuforms.git
   cd langgraph-kirokuforms
   ```

2. Install development dependencies

   ```bash
   pip install -e ".[dev]"
   ```

3. Run the offline unit tests

   ```bash
   pytest tests/test_basic.py tests/test_client_defaults.py tests/test_readme_contract.py
   ```

### Testing against a running KirokuForms

The integration tests create real tasks, and every task emails its owner. Start the KirokuForms dev server with `npm run dev:test` (which forces the console email transport), never `npm run dev`, or the suite delivers mail to real inboxes.

1. Point the tests at the server and give them a key:

   ```bash
   export KIROKU_TEST_API_KEY="your-api-key"
   export KIROKU_TEST_API_URL="http://127.0.0.1:4321/api/mcp"
   ```

2. Run the suite:

   ```bash
   pytest tests/
   ```

3. Or drive one task end to end by hand:

   ```bash
   python examples/simple_hitl_example.py
   ```

## Troubleshooting

Each of these arrives as a `ValueError` whose message names the API's error code and the HTTP status.

- **401 with a scope name in the message**: the key lacks a scope. See the scope table above.
- **429 `RATE_LIMIT_EXCEEDED`**: the daily safety rail. The client retries with backoff; the window is 24 hours. The rail sits far above a real workload, so reaching it usually means something is creating tasks in a loop.
- **429 `CONCURRENT_TASK_LIMIT_EXCEEDED`**: too many reviews left open at once. Retrying will not help until one is completed or canceled. Same caveat: this is a rail, not a quota, so meeting it is worth investigating.
- **402 `SUBSCRIPTION_REQUIRED` on assignment**: `assign_to_email` points at an address with no KirokuForms account, and the key is on a free plan. Assigning a teammate who has an account is not gated.
- **400 `VALIDATION_ERROR`**: a field is malformed. The most common cause is a non-string `defaultValue`.
- **404 on a task you just created**: the key belongs to a different account than the task.
- **409 `TASK_NOT_PENDING` on cancel**: the task is already completed, canceled, or expired.
- **The reviewer's link says the token is revoked**: the task was canceled. Create a new task.

The endpoint reference is at [kirokuforms.com/ai/mcp/tools/hitl](https://www.kirokuforms.com/ai/mcp/tools/hitl).

## License

MIT
