> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agents.kolena.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Webhook Integration

> Develop handler for webhook integration

When an Agent completes a Run, Kolena will send an HTTP request of the Run result to the webhook URL. The server listening for webhook deliveries at that URL can take action when it receives one.

### Run Result Format

When making a request to your webhook, Kolena will send a JSON object of the following format:

* `agent_id`: Agent id.
* `run_id`: Agent Run id.
* `run_url`: Link to web UI of the Run result.
* `created`: Timestamp when the Run result was created.
* `updated`: Timestamp when the Run result was last updated.
* `files`: List of files of the Run.
* `order`: User specified order of the prompts for the Agent. It is mainly used for UI display.
* `data`: Prompt results of the Run, a dictionary keyed by prompt name.

Here's an example of the JSON object:

```json theme={null}
{
  "agent_id": 1,
  "run_id": 1,
  "run_url": "https://agents.kolena.com/my_org/agents/1/runs/1",
  "created": "2025-01-01T00:00:00",
  "updated": "2025-01-05T00:00:00",
  "files": [
    "demo1.pdf",
    "demo2.csv"
  ],
  "order": [
    "prompt_1",
    "prompt_2"
  ],
  "status": "success",
  "data": {
    "prompt_1": {
      "status": "success",
      "metadata": {
        "citations": [
          {
            "file": "demo1.pdf",
            "pages": [
              2,
              5
            ]
          }
        ],
        "confidence": 1,
        "reasoning": "Looking at file demo1.pdf, ..."
      },
      "value": "foo"
    },
    "prompt_2": {
      "status": "success",
      "metadata": {
        "citations": [
          {
            "file": "demo2.csv",
            "pages": []
          }
        ],
        "confidence": 0.9,
        "reasoning": "The order amount is listed in demo2.csv"
      },
      "value": 123
    }
  }
}
```

### Respond to Webhook Request

Your webhook handler should respond to the webhook request with an HTTP 2xx status within ten seconds. We will retry up to three times if it does not, with exponential back off. If the delivery still fails after that, we will mark the delivery as failed, and will schedule another delivery after a longer period, for up to five times. If these attempts still fail, the Run result will be omitted from any further delivery.

#### Example Servers

Below are example servers demonstrating handling webhook deliveries. Since this is an example server, you can use any arbitrary value for the `KOLENA_WEBHOOK_SECRET` environment variable. When creating your own production server, you should use the secret generated by Kolena after [creating the Webhook integration](/connections#show-webhook-configuration-steps).

<Tabs>
  <Tab title="FastAPI">
    ```python main.py [expandable] theme={null}
    import logging
    import os

    from fastapi import FastAPI
    from starlette.requests import Request

    from kolena_agents import webhook

    # this is the secret you received when registering your webhook
    secret = os.environ["KOLENA_WEBHOOK_SECRET"]

    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    app = FastAPI()

    @app.post("/")
    async def root(request: Request) -> dict:
        body = await request.body()
        result = webhook.construct_event(request_body=body, secret=secret, request_headers=request.headers)
        logging.info(result.model_dump_json(indent=2))
        return dict(agent_id=result.agent_id, run_id=result.run_id)
    ```

    Start the server:

    ```
    export KOLENA_WEBHOOK_SECRET=<secret>
    fastapi run --port <port> main.py
    ```
  </Tab>

  <Tab title="Flask">
    ```python main.py [expandable] theme={null}
    import logging
    import os

    from flask import Flask
    from flask import request

    from kolena_agents import webhook

    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    app = Flask(__name__)
    secret = os.environ["KOLENA_WEBHOOK_SECRET"]

    @app.route("/", methods=["POST"])
    def root():
        body = request.get_data()
        result = webhook.construct_event(request_body=body, secret=secret, request_headers=request.headers)
        logging.info(result.model_dump_json(indent=2))
        return dict(agent_id=result.agent_id, run_id=result.run_id)
    ```

    Start the server:

    ```
    export KOLENA_WEBHOOK_SECRET=<secret>
    flask --app main.py run --port <port>
    ```
  </Tab>
</Tabs>

### Testing Webhooks

The Python SDK provides a CLI to help test your webhook locally. [Install Python SDK](/python-client/quickstart#installation) and set up an API key.

Forward events from an Agent to your webhook by using `listen` command.

```
export KOLENA_WEBHOOK_SECRET=<secret>
kolena-agents listen --agent-id <agent-id> --secret "$KOLENA_WEBHOOK_SECRET" --forward http://localhost:<port>
```

This will listen to new Agent Run results and forward them to your service set in the `--forward` option.
You can use the secret from an existing webhook integration, if you already created one, or any arbitrary value for `KOLENA_WEBHOOK_SECRET`, as we are just locally testing.
The secret passed to the CLI will be used by the `kolena_agents` CLI to sign the request forwarded to your service.
The `--tail <N>` option can be provided to fetch `N` most recent results of the Agent. If `--tail` isn't included, the command would only listen to new results.

You can test your Agent with demo data without relying on the Agent having active data using the `sample` command.

```
export KOLENA_WEBHOOK_SECRET=<secret>
kolena-agents sample --agent-id <agent-id> --secret "$KOLENA_WEBHOOK_SECRET" --forward http://localhost:<port>
```

### Register Webhook

Once your service is ready to process Kolena webhook events, follow the steps in [Integrations](/connections#webhook) to set it as a destination for your Agent.

### Technical Details

#### Kolena Custom Headers

Kolena includes additional information in request headers:

* `X-Kolena-Timestamp`: Timestamp of the request
* `X-Kolena-Delivery-ID`: Unique id for the request
* `X-Kolena-Signature`: HMAC value of request body, timestamp, and Delivery ID signed with webhook secret

You can use this information to verify that requests are coming from Kolena, detailed in the next section.

#### Signature Verification

When Kolena makes a request to your webhook URL, it will use the shared secret to create a signature from the timestamp the request is created, the request Delivery ID, and the body of the request.
The signature, timestamp and Delivery ID are put in the request headers `X-Kolena-Signature`, `X-Kolena-Timestamp`, and `X-Kolena-Delivery-ID`, respectively.
Upon receiving a request, use the secret to compute the signature from the `X-Kolena-Timestamp`, `X-Kolena-Delivery-ID` and request body, and verify the signature you computed matches the value in `X-Kolena-Signature`.
It is also advised to verify that `X-Kolena-Timestamp` is within a short time window from the time the request is received.

Kolena provides a utility function `webhook.construct_event` in our [Python SDK](/python-client/quickstart). It performs signature verification and returns the parsed JSON object.

```python theme={null}
from kolena_agents import webhook

# `secret` is the value you receive when registering the webhook
result = webhook.construct_event(request_body, secret, request_headers)
```