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:

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

Once 0.2.0 is published (see 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:

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

Quick Start

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 for the resume half.

The handler merges its outcome into the state under human_verification:

{
    "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

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

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

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

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

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.

create_verification_task

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

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:

{"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.

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

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

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)

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:

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:

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.

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

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:

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

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

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

2. Install development dependencies

   pip install -e ".[dev]"
   

3. Run the offline unit tests

   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:

   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:

   pytest tests/
   

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

   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.

License

MIT