API reference¶

Generated from the source.

The interrupt node¶

kirokuforms.graph ¶

A real LangGraph interrupt, backed by a KirokuForms review page.

Why this exists¶

create_kiroku_interrupt_handler (in kirokuforms.kirokuforms) is not a LangGraph interrupt. It creates the task and then sits in a polling loop inside the node, by default for up to an hour. The graph is not suspended during that hour; a worker is held open waiting for a human who may be asleep. It works, and it is kept for callers who depend on it, but it is a blocking REST call wearing the name of a mechanism it does not use.

LangGraph's actual mechanism is :func:langgraph.types.interrupt. It suspends the graph, persists the state through a checkpointer, and costs nothing while it waits. The run is resumed later with Command(resume=...), from a different process if you like. That is what this module provides.

The re-execution rule, which is the thing to understand¶

When a graph is resumed, the interrupted node runs again from its first line. interrupt() returns the resume value this time instead of suspending, but everything above it has already happened once and is about to happen again. A node that creates a task above its interrupt() call will create a second task on resume unless something stops it.

Nothing about that is specific to us; it catches everyone once. Here it is handled by giving each task a deterministic id derived from the graph's thread, and sending it as an idempotency key. The server has had idempotent creates since the MCP work landed: a repeat of the same key replays the case that was already made, with the same formUrl and the same token, and creates no second form, task or email. So the second execution finds the first task instead of minting one.

The consequence is worth stating: the same node interrupting twice in one thread, in a loop, would reuse one case. Pass task_id_for to make the id depend on whatever distinguishes the iterations if you need that.

create_kiroku_interrupt_node ¶

create_kiroku_interrupt_node(client: KirokuFormsHITL, *, name: str = 'review', title: str = 'Human Verification Required', description: str = 'Please verify the following information', fields: Optional[List[Dict[str, Any]]] = None, data_key: Optional[str] = None, template_id: Optional[str] = None, task_id_for: Optional[Callable[[Dict[str, Any]], str]] = None, **create_kwargs: Any) -> Callable[..., Dict[str, Any]]

Build a graph node that asks a human and suspends until they answer.

Parameters:

Name	Type	Description	Default
`client`	`KirokuFormsHITL`	A configured :class:`KirokuFormsHITL`.	required
`name`	`str`	Distinguishes this node's tasks from another node's in the same thread. It goes into the deterministic task id, so two review steps in one graph do not collide.	`'review'`
`title`	`str`	Task title, shown to the reviewer.	`'Human Verification Required'`
`description`	`str`	Instructions for the reviewer.	`'Please verify the following information'`
`fields`	`Optional[List[Dict[str, Any]]]`	Field definitions, used to build a review page for this one task.	`None`
`data_key`	`Optional[str]`	A key in the state whose dict is turned into fields by :meth:`KirokuFormsHITL.create_verification_task` (one field per entry, plus a correct/incorrect radio and a comments box).	`None`
`template_id`	`Optional[str]`	An existing form to use instead of either of the above. Prefer this for a review step that runs often. `fields` and `data_key` mint a new form per task; that form is exempt from the account's form quota, but the rows still accumulate, and a template gives every case the same page with the same field names.	`None`
`task_id_for`	`Optional[Callable[[Dict[str, Any]], str]]`	Override the deterministic id. Receives the state. Anything you return must be stable across a resume of the same logical step, because that is what stops a second task being made. Use it when one node interrupts more than once in a thread, where the default id would make both iterations share a case.	`None`
`**create_kwargs`	`Any`	Passed through to `create_task` / `create_verification_task`: `expiration`, `priority`, `callback_url`, `assign_to_email`, `assignee_name`, `assign_to_slack`.	`{}`

Returns:

Type	Description
`Callable[..., Dict[str, Any]]`	A node callable. Add it with `builder.add_node("review", node)`.

The node writes its result into state["human_verification"] as {"completed", "task_id", "form_url", "result"}, the same shape the older handler uses.

Compile the graph with a checkpointer and invoke it with a thread_id, or there is nothing to suspend into and nothing to resume::

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "order-42"}}
graph.invoke({"amount": 100}, config)          # stops at the interrupt
graph.invoke(Command(resume=answers), config)  # carries on

Source code in kirokuforms/graph.py

def create_kiroku_interrupt_node(
    client: KirokuFormsHITL,
    *,
    name: str = "review",
    title: str = "Human Verification Required",
    description: str = "Please verify the following information",
    fields: Optional[List[Dict[str, Any]]] = None,
    data_key: Optional[str] = None,
    template_id: Optional[str] = None,
    task_id_for: Optional[Callable[[Dict[str, Any]], str]] = None,
    **create_kwargs: Any,
) -> Callable[..., Dict[str, Any]]:
    """Build a graph node that asks a human and suspends until they answer.

    Args:
        client: A configured :class:`KirokuFormsHITL`.
        name: Distinguishes this node's tasks from another node's in the same
            thread. It goes into the deterministic task id, so two review steps
            in one graph do not collide.
        title: Task title, shown to the reviewer.
        description: Instructions for the reviewer.
        fields: Field definitions, used to build a review page for this one task.
        data_key: A key in the state whose dict is turned into fields by
            :meth:`KirokuFormsHITL.create_verification_task` (one field per
            entry, plus a correct/incorrect radio and a comments box).
        template_id: An existing form to use instead of either of the above.
            Prefer this for a review step that runs often. ``fields`` and
            ``data_key`` mint a new form per task; that form is exempt from the
            account's form quota, but the rows still accumulate, and a template
            gives every case the same page with the same field names.
        task_id_for: Override the deterministic id. Receives the state.
            Anything you return must be stable across a resume of the same
            logical step, because that is what stops a second task being made.
            Use it when one node interrupts more than once in a thread, where
            the default id would make both iterations share a case.
        **create_kwargs: Passed through to ``create_task`` / 
            ``create_verification_task``: ``expiration``, ``priority``,
            ``callback_url``, ``assign_to_email``, ``assignee_name``,
            ``assign_to_slack``.

    Returns:
        A node callable. Add it with ``builder.add_node("review", node)``.

    The node writes its result into ``state["human_verification"]`` as
    ``{"completed", "task_id", "form_url", "result"}``, the same shape the older
    handler uses.

    Compile the graph with a checkpointer and invoke it with a ``thread_id``, or
    there is nothing to suspend into and nothing to resume::

        graph = builder.compile(checkpointer=InMemorySaver())
        config = {"configurable": {"thread_id": "order-42"}}
        graph.invoke({"amount": 100}, config)          # stops at the interrupt
        graph.invoke(Command(resume=answers), config)  # carries on
    """
    provided = [
        name
        for name, value in (
            ("fields", fields),
            ("data_key", data_key),
            ("template_id", template_id),
        )
        if value is not None
    ]
    if not provided:
        raise ValueError("Provide fields, data_key or template_id")
    if len(provided) > 1:
        # Naming the form twice names two forms. Refusing is better than picking
        # one and silently ignoring the other.
        raise ValueError(
            f"Provide exactly one of fields, data_key or template_id; got {', '.join(provided)}"
        )

    def node(state: Dict[str, Any]) -> Dict[str, Any]:
        # Imported here, not at module scope, so `import kirokuforms` still works
        # for someone using only the REST client. The message names the fix.
        try:
            from langgraph.types import interrupt
        except ImportError as exc:  # pragma: no cover - depends on the install
            raise ImportError(
                "create_kiroku_interrupt_node needs LangGraph. "
                "Install it with `pip install langgraph`."
            ) from exc

        task_id = (
            task_id_for(state) if task_id_for else f"kf-{name}-{_thread_id()}"
        )

        # Runs on the first pass and again on every resume. It is safe to repeat
        # because task_id is stable and the server replays an idempotent create
        # rather than making a second case. See the module docstring.
        if template_id is not None:
            task = client.create_task(
                title=title,
                description=description,
                template_id=template_id,
                task_id=task_id,
                idempotency_key=task_id,
                **create_kwargs,
            )
        elif fields is not None:
            task = client.create_task(
                title=title,
                description=description,
                fields=fields,
                task_id=task_id,
                idempotency_key=task_id,
                **create_kwargs,
            )
        else:
            task = client.create_verification_task(
                title=title,
                description=description,
                data=state.get(data_key) or {},
                task_id=task_id,
                idempotency_key=task_id,
                **create_kwargs,
            )

        # Suspends here on the first pass. The value is what the caller sees in
        # the interrupt, and it carries the tokenized link, because a caller who
        # is told "a human is needed" and not told where to send them cannot do
        # anything about it.
        answers = interrupt(
            {
                "kiroku_task_id": task.get("taskId"),
                "form_url": task.get("formUrl"),
                "title": title,
                "description": description,
                "expires_at": task.get("expiresAt"),
            }
        )

        return {
            **state,
            STATE_KEY: {
                "completed": True,
                "task_id": task.get("taskId"),
                "form_url": task.get("formUrl"),
                "result": answers,
            },
        }

    node.__name__ = f"kiroku_{name}"
    return node

resume_with_answers ¶

resume_with_answers(client: KirokuFormsHITL, task_id: str, wait: bool = False) -> Command

A Command that resumes a graph with what the human actually submitted.

The alternative is to resume with whatever the operator types, which makes the form decorative. This reads the answers back from the task.

Parameters:

Name	Type	Description	Default
`client`	`KirokuFormsHITL`	The same client the node used.	required
`task_id`	`str`	The `kiroku_task_id` from the interrupt payload.	required
`wait`	`bool`	Block until the task is completed. `False` (the default) reads once and hands back a status report if nobody has answered, which is the right shape for a webhook-driven resume: you already know they answered, because that is why you were called.	`False`

Returns:

Type	Description
`Command`	`langgraph.types.Command` carrying the answers.

Source code in kirokuforms/graph.py

def resume_with_answers(
    client: KirokuFormsHITL, task_id: str, wait: bool = False
) -> "Command":
    """A ``Command`` that resumes a graph with what the human actually submitted.

    The alternative is to resume with whatever the operator types, which makes
    the form decorative. This reads the answers back from the task.

    Args:
        client: The same client the node used.
        task_id: The ``kiroku_task_id`` from the interrupt payload.
        wait: Block until the task is completed. ``False`` (the default) reads
            once and hands back a status report if nobody has answered, which is
            the right shape for a webhook-driven resume: you already know they
            answered, because that is why you were called.

    Returns:
        ``langgraph.types.Command`` carrying the answers.
    """
    try:
        from langgraph.types import Command
    except ImportError as exc:  # pragma: no cover - depends on the install
        raise ImportError(
            "resume_with_answers needs LangGraph. "
            "Install it with `pip install langgraph`."
        ) from exc

    return Command(resume=client.get_task_result(task_id, wait=wait))

The client¶

kirokuforms.kirokuforms.KirokuFormsHITL ¶

KirokuForms client for human-in-the-loop integration with LangGraph.

Source code in kirokuforms/kirokuforms.py

class KirokuFormsHITL:
    """
    KirokuForms client for human-in-the-loop integration with LangGraph.
    """

    def __init__(
        self,
        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,
    ):
        """
        Initialize the KirokuForms HITL client.

        Args:
            api_key: Your KirokuForms API key
            base_url: The KirokuForms MCP API base URL
            webhook_url: Optional URL for webhook notifications
            webhook_secret: Secret for webhook verification
            timeout: Request timeout in seconds
            max_retries: Maximum number of retries for failed requests
        """
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.webhook_url = webhook_url
        self.webhook_secret = webhook_secret
        self.timeout = timeout
        self.max_retries = max_retries

        logger.debug(f"Initializing KirokuFormsHITL client with URL: {self.base_url}")

    def _is_retryable(self, status_code: int) -> bool:
        """Whether an HTTP status is worth sending the same request again."""
        return status_code >= 500 or status_code in RETRYABLE_STATUS_CODES

    def _refusal(self, response: requests.Response) -> ValueError:
        """Build the error for a request the API refused, naming its error code.

        The API answers a refusal with ``{"success": false, "error": {"code",
        "message"}}``. A gateway or a route that does not exist can answer with
        something else entirely, so an unparseable body must still produce a
        readable error rather than a crash inside the error handling.
        """
        status = response.status_code
        try:
            body = response.json()
        except ValueError:  # requests raises a JSONDecodeError, which is a ValueError
            body = None

        error = body.get("error") if isinstance(body, dict) else None
        if isinstance(error, dict):
            code = error.get("code", f"HTTP_{status}")
            msg = error.get("message", "Unknown error")
        else:
            code = f"HTTP_{status}"
            msg = (response.text or "").strip()[:200] or "no response body"

        logger.error(f"API refused the request ({status} {code}): {msg}")
        return ValueError(f"API Error ({code}): {msg} [HTTP {status}]")

    def _request(
        self,
        method: str,
        endpoint: str,
        data: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None,
    ) -> Dict[str, Any]:
        """
        Make an API request to KirokuForms.

        Args:
            method: HTTP method (GET, POST, etc.)
            endpoint: API endpoint path (without leading slash)
            data: Optional request data

        Returns:
            API response data

        Raises:
            ValueError: If the API refuses the request (4xx, or a body reporting
                failure), naming the API's own error code.
            ConnectionError: If the transport fails, or a retryable answer (5xx,
                429) is still failing after ``max_retries``.
        """
        url = f"{self.base_url}/{endpoint}".rstrip("/")
        request_headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        if headers:
            request_headers.update(headers)

        logger.debug(f"Making {method} request to {url}")

        retries = 0
        while True:
            try:
                response = requests.request(
                    method=method,
                    url=url,
                    headers=request_headers,
                    json=data,
                    timeout=self.timeout,
                )
            except requests.exceptions.RequestException as e:
                # The request never got an answer. This is the only failure a
                # retry can fix by itself.
                if retries >= self.max_retries:
                    logger.error(
                        f"Request failed after {self.max_retries} retries: {e}"
                    )
                    raise ConnectionError(f"Failed to connect to KirokuForms API: {e}")
                retries += 1
                self._backoff(retries)
                continue

            if response.status_code >= 400:
                if self._is_retryable(response.status_code) and retries < self.max_retries:
                    retries += 1
                    logger.warning(
                        f"API answered {response.status_code}, retrying ({retries}/{self.max_retries})"
                    )
                    self._backoff(retries)
                    continue
                # A refusal is an answer, not a failed connection. Surface the
                # API's error code so the caller can act on it.
                raise self._refusal(response)

            try:
                result = response.json()
            except ValueError:
                logger.error(f"Invalid JSON response: {response.text}")
                raise ValueError(f"Invalid response from API: {response.text}")

            if not result.get("success", False):
                error = result.get("error", {})
                msg = error.get("message", "Unknown error")
                code = error.get("code", "UNKNOWN_ERROR")
                logger.error(f"API Error {code}: {msg}")
                raise ValueError(f"API Error ({code}): {msg}")

            return result.get("data", {})

    def _backoff(self, attempt: int) -> None:
        """Sleep before retrying, with a jitter so retries do not synchronize."""
        wait = 2**attempt + (time.time() % 1)
        logger.warning(f"Retrying in {wait:.2f} seconds...")
        time.sleep(wait)

    def create_task(
        self,
        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]:
        """
        Create a human-in-the-loop task.

        Either ``fields`` or ``template_id`` is required.

        Assignment forwards three keys into the MCP tool's ``settings``:

        - ``assign_to_email``: hand the task to this address. When it belongs to
          an existing KirokuForms account, the task is assigned to that account
          and the teammate is told in-app (free). When it does not, the
          recipient gets a tokenized link and needs no account, which is the
          paid feature: a key on a free plan is refused with a 402
          (``SUBSCRIPTION_REQUIRED``).
        - ``assignee_name``: an optional display name for the recipient, shown
          in the notification.
        - ``assign_to_slack``: which channel reaches the assignee. Left unset (or
          ``True``), the task is DMed on Slack when the address resolves to a
          Slack user in the owner's workspace, and emailed otherwise. ``False``
          emails a workspace teammate instead of DMing them. Slack still has to
          be connected with the direct-assign scopes and the recipient present
          in the workspace for a DM to happen, and a tokenized link is only ever
          DMed, never posted to a shared channel.

        ``idempotency_key`` is 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, instead of 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 (see
        ``kirokuforms.graph``). ``task_id`` works as one too and the server
        honours either, but the header is the one that does not also change what
        the task is called.
        """
        if fields is None and template_id is None:
            raise ValueError("Either fields or template_id must be provided")

        payload: Dict[str, Any] = {
            "title": title,
            "description": description,
            "initialData": initial_data or {},
            "settings": {
                "expiration": expiration,
                "priority": priority,
                "taskId": task_id,
                "callbackUrl": callback_url or self.webhook_url,
                "assignToEmail": assign_to_email,
                "assigneeName": assignee_name,
                # A boolean survives the None-strip below (False is not None), so
                # assign_to_slack=False reaches the wire; unset (None) is dropped
                # and the server's default Slack-when-resolvable routing applies.
                "assignToSlack": assign_to_slack,
            },
        }

        if template_id:
            payload["templateId"] = template_id
            if fields:
                payload["fields"] = fields
        elif fields:
            if not fields:
                raise ValueError(
                    "At least one field is required when not using a template"
                )
            payload["fields"] = fields

        # Remove None values
        payload = {k: v for k, v in payload.items() if v is not None}
        payload["settings"] = {
            k: v for k, v in payload["settings"].items() if v is not None
        }

        headers = (
            {"Idempotency-Key": idempotency_key} if idempotency_key else None
        )
        return self._request(
            "POST", "tools/request-human-review", payload, headers=headers
        )

    def create_verification_task(
        self,
        title: str,
        description: str,
        data: Dict[str, Any],
        fields: Optional[List[Dict[str, Any]]] = None,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        """
        Create a verification task with data to be verified.
        """
        if fields is None:
            fields = []
            for key, value in data.items():
                field_type = (
                    "radio"
                    if isinstance(value, bool)
                    else "number"
                    if isinstance(value, (int, float))
                    else "text"
                )
                base = {
                    "type": field_type,
                    "label": key.replace("_", " ").title(),
                    "name": key,
                    "required": True,
                    "defaultValue": str(value).lower()
                    if field_type == "radio"
                    else str(value),
                }
                if field_type == "radio":
                    base["options"] = [
                        {"label": "True", "value": "true"},
                        {"label": "False", "value": "false"},
                    ]
                fields.append(base)

            fields.extend(
                [
                    {
                        "type": "radio",
                        "label": "Is this information correct?",
                        "name": "is_correct",
                        "required": True,
                        "options": [
                            {"label": "Yes", "value": "yes"},
                            {"label": "No", "value": "no"},
                        ],
                    },
                    {
                        "type": "textarea",
                        "label": "Comments or Corrections",
                        "name": "comments",
                        "required": False,
                    },
                ]
            )

        return self.create_task(title, description, fields=fields, **kwargs)

    def get_task_result(
        self, task_id: str, wait: bool = True, timeout: int = 3600
    ) -> Dict[str, Any]:
        """
        Get the result (form data) of a HITL task.

        A completed task returns the submitted form data, a flat dict of field
        name to answer.

        A task that has not been completed returns a status report instead:
        ``{"taskId", "status", "completed": False, "data": None}``. That happens
        when ``wait`` is False and the human has not answered yet, and when the
        task reached a status it can never come back from (``canceled``,
        ``expired``), where waiting would only burn the timeout.

        Raises:
            TimeoutError: ``wait`` is True and ``timeout`` elapsed with the task
                still pending.
        """
        endpoint = f"resources/hitl/tasks/{task_id}"
        start = time.time()

        while True:
            result = self._request("GET", endpoint)
            status = result.get("status")

            if status == "completed":
                # Extract submission data directly from the top-level result
                submission = result.get("submission", {})
                form_data = submission.get("data", {})
                logger.debug(f"Task {task_id} completed.")
                return form_data

            if status in TERMINAL_STATUSES:
                # Canceled or expired: no human is coming, so waiting is pointless.
                logger.info(f"Task {task_id} ended as {status} without a submission.")
                return self._status_report(task_id, status)

            if not wait:
                return self._status_report(task_id, status)

            if time.time() - start >= timeout:
                raise TimeoutError(
                    f"Task {task_id} not completed within {timeout} seconds. "
                    f"Status: {status or 'unknown'}"
                )

            logger.debug(f"Task {task_id} status is {status}. Waiting...")
            time.sleep(POLL_INTERVAL_SECONDS)

    @staticmethod
    def _status_report(task_id: str, status: Optional[str]) -> Dict[str, Any]:
        """The answer for a task that has no submission to return."""
        return {
            "taskId": task_id,
            "status": status,
            "completed": False,
            "data": None,
        }

    def list_tasks(
        self,
        status: Optional[str] = None,
        limit: int = 10,
        offset: int = 0,
    ) -> Dict[str, Any]:
        """
        List HITL tasks.
        """
        params = {"limit": limit, "offset": offset}
        if status:
            params["status"] = status

        query = "&".join(f"{k}={v}" for k, v in params.items())
        endpoint = f"resources/hitl/tasks?{query}"
        return self._request("GET", endpoint)

    def verify_webhook(
        self,
        body: Any,
        signature: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Check that a webhook came from KirokuForms, using ``webhook_secret``.

        Args:
            body: The raw request body (str or bytes), or a parsed dict.
            signature: The ``X-KirokuForms-Signature-256`` header, if present.
                Falls back to the ``signature`` field inside the body.

        Returns:
            The parsed payload.

        Raises:
            WebhookVerificationError: unset secret, missing signature, or a
                digest that does not match.

        Until 2026-08-20 ``webhook_secret`` was accepted, documented as being
        for verification, and read by nothing at all. A callback endpoint built
        from the README was open to anyone who learned its URL.
        """
        from .webhooks import verify_webhook

        return verify_webhook(body, self.webhook_secret or "", signature)

    def cancel_task(self, task_id: str) -> Dict[str, Any]:
        """
        Cancel a pending HITL task.
        """
        return self._request("POST", f"resources/hitl/tasks/{task_id}/cancel")

init ¶

__init__(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)

Initialize the KirokuForms HITL client.

Parameters:

Name	Type	Description	Default
`api_key`	`str`	Your KirokuForms API key	required
`base_url`	`str`	The KirokuForms MCP API base URL	`'https://www.kirokuforms.com/api/mcp'`
`webhook_url`	`Optional[str]`	Optional URL for webhook notifications	`None`
`webhook_secret`	`Optional[str]`	Secret for webhook verification	`None`
`timeout`	`int`	Request timeout in seconds	`10`
`max_retries`	`int`	Maximum number of retries for failed requests	`3`

Source code in kirokuforms/kirokuforms.py

def __init__(
    self,
    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,
):
    """
    Initialize the KirokuForms HITL client.

    Args:
        api_key: Your KirokuForms API key
        base_url: The KirokuForms MCP API base URL
        webhook_url: Optional URL for webhook notifications
        webhook_secret: Secret for webhook verification
        timeout: Request timeout in seconds
        max_retries: Maximum number of retries for failed requests
    """
    self.api_key = api_key
    self.base_url = base_url.rstrip("/")
    self.webhook_url = webhook_url
    self.webhook_secret = webhook_secret
    self.timeout = timeout
    self.max_retries = max_retries

    logger.debug(f"Initializing KirokuFormsHITL client with URL: {self.base_url}")

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]

Create a human-in-the-loop task.

Either fields or template_id is required.

Assignment forwards three keys into the MCP tool's settings:

assign_to_email: hand the task to this address. When it belongs to an existing KirokuForms account, the task is assigned to that account and the teammate is told in-app (free). When it does not, the recipient gets a tokenized link and needs no account, which is the paid feature: a key on a free plan is refused with a 402 (SUBSCRIPTION_REQUIRED).
assignee_name: an optional display name for the recipient, shown in the notification.
assign_to_slack: which channel reaches the assignee. Left unset (or True), the task is DMed on Slack when the address resolves to a Slack user in the owner's workspace, and emailed otherwise. False emails a workspace teammate instead of DMing them. Slack still has to be connected with the direct-assign scopes and the recipient present in the workspace for a DM to happen, and a tokenized link is only ever DMed, never posted to a shared channel.

idempotency_key is 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, instead of 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 (see kirokuforms.graph). task_id works as one too and the server honours either, but the header is the one that does not also change what the task is called.

Source code in kirokuforms/kirokuforms.py

def create_task(
    self,
    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]:
    """
    Create a human-in-the-loop task.

    Either ``fields`` or ``template_id`` is required.

    Assignment forwards three keys into the MCP tool's ``settings``:

    - ``assign_to_email``: hand the task to this address. When it belongs to
      an existing KirokuForms account, the task is assigned to that account
      and the teammate is told in-app (free). When it does not, the
      recipient gets a tokenized link and needs no account, which is the
      paid feature: a key on a free plan is refused with a 402
      (``SUBSCRIPTION_REQUIRED``).
    - ``assignee_name``: an optional display name for the recipient, shown
      in the notification.
    - ``assign_to_slack``: which channel reaches the assignee. Left unset (or
      ``True``), the task is DMed on Slack when the address resolves to a
      Slack user in the owner's workspace, and emailed otherwise. ``False``
      emails a workspace teammate instead of DMing them. Slack still has to
      be connected with the direct-assign scopes and the recipient present
      in the workspace for a DM to happen, and a tokenized link is only ever
      DMed, never posted to a shared channel.

    ``idempotency_key`` is 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, instead of 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 (see
    ``kirokuforms.graph``). ``task_id`` works as one too and the server
    honours either, but the header is the one that does not also change what
    the task is called.
    """
    if fields is None and template_id is None:
        raise ValueError("Either fields or template_id must be provided")

    payload: Dict[str, Any] = {
        "title": title,
        "description": description,
        "initialData": initial_data or {},
        "settings": {
            "expiration": expiration,
            "priority": priority,
            "taskId": task_id,
            "callbackUrl": callback_url or self.webhook_url,
            "assignToEmail": assign_to_email,
            "assigneeName": assignee_name,
            # A boolean survives the None-strip below (False is not None), so
            # assign_to_slack=False reaches the wire; unset (None) is dropped
            # and the server's default Slack-when-resolvable routing applies.
            "assignToSlack": assign_to_slack,
        },
    }

    if template_id:
        payload["templateId"] = template_id
        if fields:
            payload["fields"] = fields
    elif fields:
        if not fields:
            raise ValueError(
                "At least one field is required when not using a template"
            )
        payload["fields"] = fields

    # Remove None values
    payload = {k: v for k, v in payload.items() if v is not None}
    payload["settings"] = {
        k: v for k, v in payload["settings"].items() if v is not None
    }

    headers = (
        {"Idempotency-Key": idempotency_key} if idempotency_key else None
    )
    return self._request(
        "POST", "tools/request-human-review", payload, headers=headers
    )

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]

Create a verification task with data to be verified.

Source code in kirokuforms/kirokuforms.py

def create_verification_task(
    self,
    title: str,
    description: str,
    data: Dict[str, Any],
    fields: Optional[List[Dict[str, Any]]] = None,
    **kwargs: Any,
) -> Dict[str, Any]:
    """
    Create a verification task with data to be verified.
    """
    if fields is None:
        fields = []
        for key, value in data.items():
            field_type = (
                "radio"
                if isinstance(value, bool)
                else "number"
                if isinstance(value, (int, float))
                else "text"
            )
            base = {
                "type": field_type,
                "label": key.replace("_", " ").title(),
                "name": key,
                "required": True,
                "defaultValue": str(value).lower()
                if field_type == "radio"
                else str(value),
            }
            if field_type == "radio":
                base["options"] = [
                    {"label": "True", "value": "true"},
                    {"label": "False", "value": "false"},
                ]
            fields.append(base)

        fields.extend(
            [
                {
                    "type": "radio",
                    "label": "Is this information correct?",
                    "name": "is_correct",
                    "required": True,
                    "options": [
                        {"label": "Yes", "value": "yes"},
                        {"label": "No", "value": "no"},
                    ],
                },
                {
                    "type": "textarea",
                    "label": "Comments or Corrections",
                    "name": "comments",
                    "required": False,
                },
            ]
        )

    return self.create_task(title, description, fields=fields, **kwargs)

get_task_result ¶

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

Get the result (form data) of a HITL task.

A completed task returns the submitted form data, a flat dict of field name to answer.

A task that has not been completed returns a status report instead: {"taskId", "status", "completed": False, "data": None}. That happens when wait is False and the human has not answered yet, and when the task reached a status it can never come back from (canceled, expired), where waiting would only burn the timeout.

Raises:

Type	Description
`TimeoutError`	`wait` is True and `timeout` elapsed with the task still pending.

Source code in kirokuforms/kirokuforms.py

def get_task_result(
    self, task_id: str, wait: bool = True, timeout: int = 3600
) -> Dict[str, Any]:
    """
    Get the result (form data) of a HITL task.

    A completed task returns the submitted form data, a flat dict of field
    name to answer.

    A task that has not been completed returns a status report instead:
    ``{"taskId", "status", "completed": False, "data": None}``. That happens
    when ``wait`` is False and the human has not answered yet, and when the
    task reached a status it can never come back from (``canceled``,
    ``expired``), where waiting would only burn the timeout.

    Raises:
        TimeoutError: ``wait`` is True and ``timeout`` elapsed with the task
            still pending.
    """
    endpoint = f"resources/hitl/tasks/{task_id}"
    start = time.time()

    while True:
        result = self._request("GET", endpoint)
        status = result.get("status")

        if status == "completed":
            # Extract submission data directly from the top-level result
            submission = result.get("submission", {})
            form_data = submission.get("data", {})
            logger.debug(f"Task {task_id} completed.")
            return form_data

        if status in TERMINAL_STATUSES:
            # Canceled or expired: no human is coming, so waiting is pointless.
            logger.info(f"Task {task_id} ended as {status} without a submission.")
            return self._status_report(task_id, status)

        if not wait:
            return self._status_report(task_id, status)

        if time.time() - start >= timeout:
            raise TimeoutError(
                f"Task {task_id} not completed within {timeout} seconds. "
                f"Status: {status or 'unknown'}"
            )

        logger.debug(f"Task {task_id} status is {status}. Waiting...")
        time.sleep(POLL_INTERVAL_SECONDS)

list_tasks ¶

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

List HITL tasks.

Source code in kirokuforms/kirokuforms.py

def list_tasks(
    self,
    status: Optional[str] = None,
    limit: int = 10,
    offset: int = 0,
) -> Dict[str, Any]:
    """
    List HITL tasks.
    """
    params = {"limit": limit, "offset": offset}
    if status:
        params["status"] = status

    query = "&".join(f"{k}={v}" for k, v in params.items())
    endpoint = f"resources/hitl/tasks?{query}"
    return self._request("GET", endpoint)

verify_webhook ¶

verify_webhook(body: Any, signature: Optional[str] = None) -> Dict[str, Any]

Check that a webhook came from KirokuForms, using webhook_secret.

Parameters:

Name	Type	Description	Default
`body`	`Any`	The raw request body (str or bytes), or a parsed dict.	required
`signature`	`Optional[str]`	The `X-KirokuForms-Signature-256` header, if present. Falls back to the `signature` field inside the body.	`None`

Returns:

Type	Description
`Dict[str, Any]`	The parsed payload.

Raises:

Type	Description
`WebhookVerificationError`	unset secret, missing signature, or a digest that does not match.

Until 2026-08-20 webhook_secret was accepted, documented as being for verification, and read by nothing at all. A callback endpoint built from the README was open to anyone who learned its URL.

Source code in kirokuforms/kirokuforms.py

def verify_webhook(
    self,
    body: Any,
    signature: Optional[str] = None,
) -> Dict[str, Any]:
    """Check that a webhook came from KirokuForms, using ``webhook_secret``.

    Args:
        body: The raw request body (str or bytes), or a parsed dict.
        signature: The ``X-KirokuForms-Signature-256`` header, if present.
            Falls back to the ``signature`` field inside the body.

    Returns:
        The parsed payload.

    Raises:
        WebhookVerificationError: unset secret, missing signature, or a
            digest that does not match.

    Until 2026-08-20 ``webhook_secret`` was accepted, documented as being
    for verification, and read by nothing at all. A callback endpoint built
    from the README was open to anyone who learned its URL.
    """
    from .webhooks import verify_webhook

    return verify_webhook(body, self.webhook_secret or "", signature)

cancel_task ¶

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

Cancel a pending HITL task.

Source code in kirokuforms/kirokuforms.py

def cancel_task(self, task_id: str) -> Dict[str, Any]:
    """
    Cancel a pending HITL task.
    """
    return self._request("POST", f"resources/hitl/tasks/{task_id}/cancel")

Webhook verification¶

kirokuforms.webhooks ¶

Verifying that a webhook really came from KirokuForms.

KirokuFormsHITL has accepted a webhook_secret argument since the first release, documented as "Secret for webhook verification". It was stored on the instance and never read: nothing verified anything. A caller following the README believed their callback endpoint was authenticated when it was open to anyone who learned the URL. This module is the missing half.

How KirokuForms signs¶

src/lib/services/webhook-service.ts computes::

HMAC-SHA256(secret, JSON.stringify(payload without its "signature" key))

hex-encoded, and sends it two ways: in the X-KirokuForms-Signature-256 header, and as a signature field inside the JSON body it delivers.

That means the signed bytes are not the bytes on the wire: the delivered body has the signature added to it. A verifier has to parse the JSON, drop signature, and re-serialize exactly the way JSON.stringify would.

Re-serializing is the fragile part, and it is fragile in both languages, so it is worth being explicit about what makes it work here:

separators=(",", ":") matches JSON.stringify, which inserts no spaces.
ensure_ascii=False matches it too: JavaScript emits real UTF-8 for non-ASCII, Python escapes it to \uXXXX unless told otherwise.
Key order is preserved. json.loads fills a dict in document order, Python dicts keep insertion order, and JavaScript's {signature, ...rest} rest spread keeps the order of everything it copies. So both sides walk the keys in the order they arrived on the wire.
Whole numbers stay whole. JavaScript has one number type and never writes 1.0, so a whole number arrives as 1, and json.loads makes that an int, which json.dumps writes back as 1.

tests/test_webhooks.py checks this against signatures generated by the actual TypeScript computeSignature, not against a Python reimplementation agreeing with itself.

WebhookVerificationError ¶

Bases: Exception

The payload did not come from KirokuForms, or was altered on the way.

Source code in kirokuforms/webhooks.py

class WebhookVerificationError(Exception):
    """The payload did not come from KirokuForms, or was altered on the way."""

verify_webhook ¶

verify_webhook(body: Union[str, bytes, Dict[str, Any]], secret: str, signature: Optional[str] = None) -> Dict[str, Any]

Check a webhook's signature and return its parsed payload.

Parameters:

Name	Type	Description	Default
`body`	`Union[str, bytes, Dict[str, Any]]`	The raw request body, or an already-parsed dict.	required
`secret`	`str`	The webhook secret configured for this endpoint.	required
`signature`	`Optional[str]`	The `X-KirokuForms-Signature-256` header, if you have it. When omitted, the `signature` field inside the body is used. Prefer the header: it is the copy an attacker cannot simply recompute after editing the body, because they do not have the secret either way, but keeping them separate means a proxy that rewrites the body cannot quietly re-sign it.	`None`

Returns:

Type	Description
`Dict[str, Any]`	The payload as a dict, with `signature` still on it.

Raises:

Type	Description
`WebhookVerificationError`	no secret, no signature, unparseable body, or a digest that does not match.

Compared with :func:hmac.compare_digest, so a wrong signature takes the same time to reject as a nearly-right one.

Source code in kirokuforms/webhooks.py

def verify_webhook(
    body: Union[str, bytes, Dict[str, Any]],
    secret: str,
    signature: Optional[str] = None,
) -> Dict[str, Any]:
    """Check a webhook's signature and return its parsed payload.

    Args:
        body: The raw request body, or an already-parsed dict.
        secret: The webhook secret configured for this endpoint.
        signature: The ``X-KirokuForms-Signature-256`` header, if you have it.
            When omitted, the ``signature`` field inside the body is used.
            Prefer the header: it is the copy an attacker cannot simply
            recompute after editing the body, because they do not have the
            secret either way, but keeping them separate means a proxy that
            rewrites the body cannot quietly re-sign it.

    Returns:
        The payload as a dict, with ``signature`` still on it.

    Raises:
        WebhookVerificationError: no secret, no signature, unparseable body, or
            a digest that does not match.

    Compared with :func:`hmac.compare_digest`, so a wrong signature takes the
    same time to reject as a nearly-right one.
    """
    if not secret:
        raise WebhookVerificationError(
            "No webhook secret. Pass the same secret you configured on the "
            "webhook in KirokuForms."
        )

    if isinstance(body, (str, bytes)):
        text = body.decode("utf-8") if isinstance(body, bytes) else body
        try:
            payload = json.loads(text)
        except ValueError as exc:
            raise WebhookVerificationError(
                f"Webhook body is not JSON: {exc}"
            ) from exc
    else:
        payload = body

    if not isinstance(payload, dict):
        raise WebhookVerificationError(
            f"Webhook body is {type(payload).__name__}, expected a JSON object."
        )

    provided = signature or payload.get("signature")
    if not provided:
        raise WebhookVerificationError(
            f"No signature. Pass the {SIGNATURE_HEADER} header, or send a body "
            "that carries a `signature` field. KirokuForms only signs when the "
            "webhook has a secret configured, so an unsigned delivery usually "
            "means the secret is set on your side and not on the webhook."
        )

    expected = compute_signature(payload, secret)
    if not hmac.compare_digest(str(provided), expected):
        raise WebhookVerificationError(
            "Webhook signature does not match. Either the secret is wrong, or "
            "the body was modified after it was signed."
        )

    return payload

compute_signature ¶

compute_signature(payload: Dict[str, Any], secret: str) -> str

The signature KirokuForms would send for this payload.

Source code in kirokuforms/webhooks.py

def compute_signature(payload: Dict[str, Any], secret: str) -> str:
    """The signature KirokuForms would send for this payload."""
    return hmac.new(
        secret.encode("utf-8"),
        canonical_payload(payload).encode("utf-8"),
        sha256,
    ).hexdigest()

canonical_payload ¶

canonical_payload(payload: Dict[str, Any]) -> str

The exact string KirokuForms ran the HMAC over, for a parsed payload.

Source code in kirokuforms/webhooks.py

def canonical_payload(payload: Dict[str, Any]) -> str:
    """The exact string KirokuForms ran the HMAC over, for a parsed payload."""
    unsigned = {k: v for k, v in payload.items() if k != "signature"}
    return json.dumps(unsigned, separators=(",", ":"), ensure_ascii=False)

Deprecated¶

kirokuforms.kirokuforms.create_kiroku_interrupt_handler ¶

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

Create a blocking human-review step. Deprecated since 0.3.0.

Use :func:kirokuforms.create_kiroku_interrupt_node instead.

This is not a LangGraph interrupt, despite the name. It creates the task and then polls in a loop inside the node, by default for up to an hour, holding a worker open the whole time. The graph is never suspended: nothing is checkpointed, nothing can be resumed in another process, and a reviewer who goes to lunch costs you a live worker until they come back.

create_kiroku_interrupt_node uses LangGraph's own :func:interrupt. The graph suspends, the state is checkpointed, the process is free, and the run resumes later from anywhere with Command(resume=...). It also sends an idempotency key, so the re-execution that happens on every resume finds the case it already made instead of mailing your reviewer a second copy.

Kept, and still working, because published code imports it. It will not gain features. Migration is usually four lines::

# before
handler = create_kiroku_interrupt_handler(api_key="...")
def node(state):
    return handler(state, {"title": "Approve?", "fields": FIELDS})

# after
node = create_kiroku_interrupt_node(
    KirokuFormsHITL(api_key="..."), name="approval",
    title="Approve?", fields=FIELDS,
)
# then compile the graph with a checkpointer and invoke with a thread_id

Both write their outcome to state["human_verification"], so nothing downstream of the node has to change.

Source code in kirokuforms/kirokuforms.py

def create_kiroku_interrupt_handler(
    api_key: str, **kwargs: Any
) -> Callable[[Dict[str, Any], Dict[str, Any]], Dict[str, Any]]:
    """Create a blocking human-review step. **Deprecated since 0.3.0.**

    Use :func:`kirokuforms.create_kiroku_interrupt_node` instead.

    This is not a LangGraph interrupt, despite the name. It creates the task and
    then polls in a loop *inside the node*, by default for up to an hour, holding
    a worker open the whole time. The graph is never suspended: nothing is
    checkpointed, nothing can be resumed in another process, and a reviewer who
    goes to lunch costs you a live worker until they come back.

    ``create_kiroku_interrupt_node`` uses LangGraph's own :func:`interrupt`. The
    graph suspends, the state is checkpointed, the process is free, and the run
    resumes later from anywhere with ``Command(resume=...)``. It also sends an
    idempotency key, so the re-execution that happens on every resume finds the
    case it already made instead of mailing your reviewer a second copy.

    Kept, and still working, because published code imports it. It will not gain
    features. Migration is usually four lines::

        # before
        handler = create_kiroku_interrupt_handler(api_key="...")
        def node(state):
            return handler(state, {"title": "Approve?", "fields": FIELDS})

        # after
        node = create_kiroku_interrupt_node(
            KirokuFormsHITL(api_key="..."), name="approval",
            title="Approve?", fields=FIELDS,
        )
        # then compile the graph with a checkpointer and invoke with a thread_id

    Both write their outcome to ``state["human_verification"]``, so nothing
    downstream of the node has to change.
    """
    warnings.warn(
        "create_kiroku_interrupt_handler is deprecated since 0.3.0 and will not "
        "gain features: it polls inside the node instead of suspending the "
        "graph, so a worker is held open for as long as the human takes. Use "
        "create_kiroku_interrupt_node, which uses LangGraph's interrupt() and "
        "checkpointing. See https://www.kirokuforms.com/ai/langgraph",
        DeprecationWarning,
        stacklevel=2,
    )
    client = KirokuFormsHITL(api_key, **kwargs)

    def interrupt_handler(
        state: Dict[str, Any], interrupt_data: Dict[str, Any]
    ) -> Dict[str, Any]:
        title = interrupt_data.get("title", "Human Verification Required")
        description = interrupt_data.get(
            "description", "Please verify the following information"
        )
        fields = interrupt_data.get("fields", [])
        data = interrupt_data.get("data", {})
        wait_for_result = interrupt_data.get("wait_for_result", True)

        if not fields and data:
            task = client.create_verification_task(
                title=title,
                description=description,
                data=data,
            )
        else:
            task = client.create_task(
                title=title,
                description=description,
                fields=fields,
            )

        result = {
            **state,
            "human_verification": {
                "completed": False,
                "task_id": task.get("taskId"),
                "form_url": task.get("formUrl"),
                "result": None,
            },
        }

        if wait_for_result:
            try:
                submission = client.get_task_result(task["taskId"])
            except TimeoutError:
                logger.warning(f"Task {task['taskId']} not completed within timeout")
            else:
                # A task that was canceled or expired comes back as a status
                # report, not as answers. Filing that under "result" would tell
                # the graph a human approved something nobody ever looked at.
                if submission.get("completed") is False:
                    logger.warning(
                        f"Task {task['taskId']} ended as "
                        f"{submission.get('status')} without a submission"
                    )
                else:
                    result["human_verification"]["completed"] = True
                    result["human_verification"]["result"] = submission

        return result

    return interrupt_handler

API reference¶

The interrupt node¶

kirokuforms.graph ¶

Why this exists¶

The re-execution rule, which is the thing to understand¶

create_kiroku_interrupt_node ¶

resume_with_answers ¶

The client¶

kirokuforms.kirokuforms.KirokuFormsHITL ¶

__init__ ¶

create_task ¶

create_verification_task ¶

get_task_result ¶

list_tasks ¶

verify_webhook ¶

cancel_task ¶

Webhook verification¶

kirokuforms.webhooks ¶

How KirokuForms signs¶

WebhookVerificationError ¶

verify_webhook ¶

compute_signature ¶

canonical_payload ¶

Deprecated¶

kirokuforms.kirokuforms.create_kiroku_interrupt_handler ¶

init ¶