# Kovasign Document API — Python

The Document API is headless parity with the Kovasign platform: whatever a
sender can do by clicking through the app, an Enterprise tenant's own code
can do by calling this API. It requires an Enterprise plan and a Document
API token minted in Workspace → Developers → API & webhooks ("Connect your
software" there hands you all of these values as a ready .env block).

- Base URL: `https://app.kovasign.com/api/v1` (`KOVASIGN_API_URL`)
- Every `/documents*` and `/templates*` call needs three headers:
  - `Authorization: Bearer kvs_doc_…` (`KOVASIGN_API_TOKEN`)
  - `X-On-Behalf-Of: <member email or profile uuid>` (`KOVASIGN_ON_BEHALF_OF`) — the
    real person the action is attributed to. **Not** sent on `/webhooks*`.
  - `Idempotency-Key: <string ≤128 chars>` — required on `POST /documents`,
    `POST /documents/{id}/send`, and `POST /webhooks/{id}/test`; optional
    (but recommended) on every other write.
    **One key per attempt, not per record:** generate a fresh key when you start
    an action, store it with the exact body (including `confirmed_at`), and retry
    a timeout/5xx with the same key and the byte-identical body. Any new intent
    (re-send after a recall, a new document after a void) needs a new key — an
    old key replays the old response, or 422s `idempotency_key_reused` if the
    body changed. Keys are remembered for at least 24 hours.
  - Sign-then-send needs the company's role to have the lowest `signing_order`
    (Acme HR is step 1 below). Signing a draft emails nobody; `/send` invites the
    first step that still has an unsigned signer.

Only `requests`, `hashlib`, `hmac`, `time`, `json`, and `http.server` are
used below.

```python
import os
import requests

KOVASIGN_API_URL = os.environ.get("KOVASIGN_API_URL", "https://app.kovasign.com/api/v1")
KOVASIGN_API_TOKEN = os.environ["KOVASIGN_API_TOKEN"]
KOVASIGN_ON_BEHALF_OF = os.environ["KOVASIGN_ON_BEHALF_OF"]


def auth_headers(idempotency_key: str | None = None) -> dict:
    """Shared headers for every /documents* and /templates* call."""
    headers = {
        "Authorization": f"Bearer {KOVASIGN_API_TOKEN}",
        "X-On-Behalf-Of": KOVASIGN_ON_BEHALF_OF,
    }
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key
    return headers
```

## 1. Create a document with placement tags

Upload a PDF whose text already contains `{{sig:signer1}} {{date:signer1}} {{sig:signer2}} {{text:signer2:title}}`, map the two roles, and place fields from the tags — `multipart/form-data` with a `file` part and a `meta` JSON part.

```python
import json


def create_document_from_file(pdf_path: str) -> dict:
    meta = {
        "name": "Offer — J. Tan",
        "roles": {
            "signer1": {"name": "Jane Tan", "email": "jane@x.com", "signing_order": 2},
            "signer2": {"name": "Acme HR", "email": "hr@acme.com", "signing_order": 1},
        },
        "placement": {"tags": True},
        "expires_in_days": 7,
    }
    with open(pdf_path, "rb") as f:
        resp = requests.post(
            f"{KOVASIGN_API_URL}/documents",
            headers=auth_headers("offer-8812"),
            files={"file": ("offer-letter.pdf", f, "application/pdf")},
            data={"meta": json.dumps(meta)},
        )

    if resp.status_code == 422:
        body = resp.json()
        # No document was created — the draft is rolled back on any unresolved field.
        reasons = ", ".join(e["reason"] for e in body["errors"])
        raise RuntimeError(f"placement_unresolved: {reasons}")
    resp.raise_for_status()
    return resp.json()
```

A `201` looks like:

```json
{
  "document": { "id": "d1", "name": "Offer — J. Tan", "status": "draft" },
  "recipients": [{ "id": "r1", "role": "signer1", "name": "Jane Tan", "email": "jane@x.com", "type": "signer", "signing_order": 2, "status": "pending" }],
  "placed": [{ "field_id": "f1", "recipient_id": "r1", "type": "signature", "page": 3, "x": 12.4, "y": 71.0, "width": 15, "height": 5.2, "via": "tag", "tag": "{{sig:signer1}}" }],
  "warnings": []
}
```

## 2. Create from a URL or from a template

From a public HTTPS URL (SSRF-checked) instead of an upload:

```python
def create_document_from_url() -> dict:
    body = {
        "source": {"url": "https://cdn.acme.com/offer.pdf"},
        "name": "Offer — J. Tan",
        "roles": {
            "signer1": {"name": "Jane Tan", "email": "jane@x.com", "signing_order": 2},
            "signer2": {"name": "Acme HR", "email": "hr@acme.com", "signing_order": 1},
        },
        "placement": {"tags": True},
        "expires_in_days": 7,
    }
    resp = requests.post(f"{KOVASIGN_API_URL}/documents", headers=auth_headers("offer-8812-url"), json=body)
    resp.raise_for_status()
    return resp.json()
```

From a saved template — only the roles you name are overridden; the rest keep the template's remembered name/email:

```python
def create_document_from_template() -> dict:
    body = {
        "template_id": "tmpl_1",
        "name": "Offer — J. Tan",
        "roles": {"signer1": {"name": "Jane Tan", "email": "jane@x.com"}},
    }
    resp = requests.post(f"{KOVASIGN_API_URL}/documents", headers=auth_headers("offer-8812-tmpl"), json=body)
    resp.raise_for_status()
    return resp.json()
```

## 3. Sender signs first

Fetch the document, find the sender's own date field, sign it, and fill the field in the same call.

```python
from datetime import datetime, timezone


def sender_signs_first(document_id: str, sender_recipient_id: str) -> dict:
    get_resp = requests.get(f"{KOVASIGN_API_URL}/documents/{document_id}", headers=auth_headers())
    get_resp.raise_for_status()
    fields = get_resp.json()["fields"]

    date_field = next(
        (f for f in fields if f["recipient_id"] == sender_recipient_id and f["field_type"] == "date"),
        None,
    )
    if date_field is None:
        raise RuntimeError("sender date field not found")

    body = {
        "intent": {
            "user_confirmed": True,
            "confirmed_at": datetime.now(timezone.utc).isoformat(),
            "surface": "acme-hris/v4",
            "actor_ref": KOVASIGN_ON_BEHALF_OF,
        },
        "field_values": {date_field["id"]: datetime.now(timezone.utc).date().isoformat()},
    }
    # A Singpass-required slot answers 409 singpass_required — it can only be
    # signed by that recipient via their own link, never headlessly.
    sign_resp = requests.post(f"{KOVASIGN_API_URL}/documents/{document_id}/sign", headers=auth_headers(), json=body)
    sign_resp.raise_for_status()
    return sign_resp.json()
```

## 4. Send

```python
# `attempt_key` / `confirmed_at` are created ONCE per send attempt and stored
# before the first call (e.g. attempt_key = str(uuid.uuid4()), confirmed_at =
# when the human approved). Retry a timeout/5xx with the same pair so the body
# is byte-identical; a re-send after a recall is a NEW attempt with a new key.
def send_document(document_id: str, attempt_key: str, confirmed_at: str) -> dict:
    body = {
        "attestation": {
            "user_confirmed": True,
            "confirmed_at": confirmed_at,
            "surface": "acme-hris/v4",
        }
    }
    resp = requests.post(
        f"{KOVASIGN_API_URL}/documents/{document_id}/send",
        headers=auth_headers(attempt_key),
        json=body,
    )
    resp.raise_for_status()
    return resp.json()
```

## 5. Poll for completion (and the webhook alternative)

```python
import time


def wait_for_completion(document_id: str, poll_seconds: float = 5.0) -> None:
    while True:
        resp = requests.get(f"{KOVASIGN_API_URL}/documents/{document_id}", headers=auth_headers())
        resp.raise_for_status()
        if resp.json()["document"]["status"] == "completed":
            return
        time.sleep(poll_seconds)
```

Polling works, but it's simpler and faster to register a webhook once and react to `document.completed` as it happens — see §8 below for a full receiver.

## 6. Download the signed PDF and certificate

Bytes, written to disk, verified against `X-Content-SHA256`:

```python
import hashlib


def download_signed_pdf(document_id: str, out_path: str) -> None:
    resp = requests.get(f"{KOVASIGN_API_URL}/documents/{document_id}/files/signed", headers=auth_headers())
    resp.raise_for_status()  # 409 not_completed until every signer has signed
    expected_sha256 = resp.headers.get("X-Content-SHA256")
    content = resp.content
    actual_sha256 = hashlib.sha256(content).hexdigest()
    if expected_sha256 and expected_sha256 != actual_sha256:
        raise RuntimeError("sha256 mismatch")
    with open(out_path, "wb") as f:
        f.write(content)
```

Or the URL variant, which also returns the hash to verify against once you fetch the bytes:

```python
def get_signed_file_url(document_id: str) -> dict:
    resp = requests.get(
        f"{KOVASIGN_API_URL}/documents/{document_id}/files/signed",
        headers=auth_headers(),
        params={"disposition": "url"},
    )
    resp.raise_for_status()
    return resp.json()  # {"url": "...", "expires_in": 3600, "sha256": "..."}
```

The Signing Log Certificate uses the same two shapes at `/files/certificate`:

```python
def download_certificate(document_id: str, out_path: str) -> None:
    resp = requests.get(f"{KOVASIGN_API_URL}/documents/{document_id}/files/certificate", headers=auth_headers())
    resp.raise_for_status()
    with open(out_path, "wb") as f:
        f.write(resp.content)
```

## 7. Templates and voiding

```python
def save_as_template(document_id: str, name: str) -> dict:
    resp = requests.post(
        f"{KOVASIGN_API_URL}/documents/{document_id}/save-as-template",
        headers=auth_headers(),
        json={"name": name},
    )
    resp.raise_for_status()
    return resp.json()


def find_template_by_name(search: str) -> dict | None:
    resp = requests.get(f"{KOVASIGN_API_URL}/templates", headers=auth_headers(), params={"search": search})
    resp.raise_for_status()
    templates = resp.json()["templates"]
    return templates[0] if templates else None


def create_from_template(template_id: str) -> dict:
    """Create a document from a template, overriding one prefilled role."""
    body = {
        "template_id": template_id,
        "name": "Offer — K. Lim",
        "roles": {"signer1": {"name": "Kevin Lim", "email": "kevin@x.com"}},
    }
    resp = requests.post(f"{KOVASIGN_API_URL}/documents", headers=auth_headers("offer-9001"), json=body)
    resp.raise_for_status()
    return resp.json()


def void_document(document_id: str, reason: str) -> None:
    resp = requests.post(
        f"{KOVASIGN_API_URL}/documents/{document_id}/void",
        headers=auth_headers(),
        json={"reason": reason},
    )
    resp.raise_for_status()
```

## 8. Webhook receiver — verify `Kovasign-Signature`

Port of the reference verifier in `docs/runbooks/tenant-seat-webhooks-fixtures.md` §4
(shared with `docs/runbooks/document-webhooks.md` §2/§4). Rules: parse
`t=…,v1=…[,v1=…]`; reject if `|now − t| > 300s`; HMAC-SHA256 over
`"<t>.<raw body>"`; constant-time compare; accept if **any** `v1` matches
**any** secret you currently hold (both terms are valid for 24h after a
rotation); dedupe on `event_id`; apply an event only when its `sequence` is
greater than the last one applied for that `document_id`; always ack `200`
for a duplicate or an out-of-order event, otherwise Kovasign retries it.

```python
import hashlib
import hmac
import json
import os
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

TOLERANCE_SEC = 300  # 5-minute replay window

# Held secrets for this endpoint — both current and (during a rotation) previous.
WEBHOOK_SECRETS = [s for s in [os.environ.get("KOVASIGN_WEBHOOK_SECRET")] if s]

# In-memory dedupe + ordering state. Use a real DB table in production
# (event_id PRIMARY KEY; last_sequence per document_id).
_seen_event_ids: set[str] = set()
_last_sequence_by_document: dict[str, int] = {}


def verify_kovasign_signature(
    raw_body: bytes,
    signature_header: str,
    secrets: list[str],
    now_sec: float | None = None,
) -> bool:
    """Verbatim port of the reference verifier — verifies over the RAW body, never a re-serialised parse."""
    now_sec = time.time() if now_sec is None else now_sec

    # signature_header: "t=1758700502,v1=<hex>[,v1=<hex>]"
    parts = [p.split("=", 1) for p in signature_header.split(",")]
    t = next((v for k, v in parts if k == "t"), None)
    v1s = [v for k, v in parts if k == "v1"]
    if t is None or not v1s:
        return False
    if abs(now_sec - int(t)) > TOLERANCE_SEC:
        return False

    signed = f"{t}.".encode() + raw_body
    for secret in secrets:
        expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
        for v1 in v1s:
            if hmac.compare_digest(v1, expected):
                return True
    return False


class WebhookHandler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        length = int(self.headers.get("Content-Length", "0"))
        raw_body = self.rfile.read(length)
        signature_header = self.headers.get("Kovasign-Signature")

        if not signature_header or not verify_kovasign_signature(raw_body, signature_header, WEBHOOK_SECRETS):
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b"invalid signature")
            return

        event = json.loads(raw_body)

        # Dedupe on event_id — always ack 200 for a duplicate so Kovasign stops retrying it.
        if event["id"] in _seen_event_ids:
            self._ack(b"duplicate")
            return
        _seen_event_ids.add(event["id"])

        # Order on sequence, never on occurred_at. Apply only when strictly
        # greater than the last sequence applied for this document; still ack 200 otherwise.
        document_id = event.get("document_id")
        if document_id:
            last_sequence = _last_sequence_by_document.get(document_id, -1)
            if event["sequence"] <= last_sequence:
                self._ack(b"out of order, discarded")
                return
            _last_sequence_by_document[document_id] = event["sequence"]

        # Apply the event — e.g. react to document.completed by downloading the signed PDF.
        print(f"applied {event['type']} for document {document_id} (sequence {event['sequence']})")
        self._ack(b"ok")

    def _ack(self, body: bytes) -> None:
        self.send_response(200)
        self.end_headers()
        self.wfile.write(body)


if __name__ == "__main__":
    server = HTTPServer(("0.0.0.0", 3001), WebhookHandler)
    print("webhook receiver listening on :3001")
    server.serve_forever()
```