Write a push source

​

Prerequisites

• Python 3.12+ (python3 --version), a Quivr checkout and its CLI (go build -o .scratch/quivr ./cmd/quivr). Add .scratch to your PATH.
• Read How plugins work for the engine/plugin boundary and Write a connector for source kinds and stable record identities.
• For signature routes, the provider’s signature format and signing secret. The operator deposits that secret as the instance’s credential; you receive it decrypted in the handler. The offline sample below needs no provider account.

​

Steps

​

Generate a push source

quivr plugin init incoming-records --kind connector --push
python3 -m venv incoming-records/.venv
. incoming-records/.venv/bin/activate
pip install -e sdks/python
cd incoming-records
python3 -m unittest discover -s tests
quivr plugin test --report contract-report.json

​

Declare and handle a route

- name: publish
  method: POST
  path: records
  auth: instance_token
  request_schema:
    type: object
    additionalProperties: false
    required: [key, revision, title, text]
    properties:
      key: {type: string, minLength: 1, maxLength: 256}
      revision: {type: string, minLength: 1, maxLength: 256}
      title: {type: string, minLength: 1, maxLength: 256}
      text: {type: string, minLength: 1, maxLength: 65536}

​

Choose authentication

​

Verify provider signatures

​

Answer provider challenges

- name: receive
  method: POST
  path: receive
  auth: signature
  signature: {header: x-twitter-webhooks-signature, window_seconds: 300}
- name: challenge
  method: GET
  path: receive
  auth: signature
  signature: {header: x-twitter-webhooks-signature, window_seconds: 300}

import base64
import hashlib
import hmac
import json
from urllib.parse import parse_qs

from quivr_plugin import ConnectorReceiveResponse, ReceiveAnswer, ReceiveRequest


@plugin.connector_route("events", "receive")
@plugin.connector_route("events", "challenge")
def receive(request: ReceiveRequest) -> ConnectorReceiveResponse:
    secret = request.credential.decode()["consumer_secret"].encode()

    def sign(raw: bytes) -> str:
        digest = hmac.new(secret, raw, hashlib.sha256).digest()
        return "sha256=" + base64.b64encode(digest).decode("ascii")

    if request.request.method == "GET":
        token = parse_qs(request.request.query).get("crc_token", [""])[0]
        if not token:
            return ConnectorReceiveResponse(
                verdict="refused", response=ReceiveAnswer(status=400, body="crc_token is missing")
            )
        return ConnectorReceiveResponse(
            verdict="accepted",
            response=ReceiveAnswer(
                status=200, content_type="application/json",
                body=json.dumps({"response_token": sign(token.encode())}),
            ),
        )

    raw = base64.b64decode(request.request.body_base64)
    values = request.request.headers.get("x-twitter-webhooks-signature", [])
    if len(values) != 1 or not hmac.compare_digest(values[0], sign(raw)):
        return ConnectorReceiveResponse(
            verdict="refused", response=ReceiveAnswer(status=401, body="invalid signature")
        )
    return ConnectorReceiveResponse(
        verdict="accepted", response=ReceiveAnswer(status=202), items=[]
    )

​

Push checks

1. Quivr recognizes a supplied bearer, checks a recognized API key’s push permission and bounds the request.
2. It loads the enabled instance, checks API key Organization and Corpus scope, resolves the route and checks its authentication, including an instance token.
3. Signed POSTs require one signature header, check any timestamp’s age and reserve signature and delivery-key fingerprints to refuse replays. GET challenges skip these guards.
4. Quivr checks delivery-key metadata, parses JSON and validates the route schema, then checks the IP allowlist.
5. A completed bearer cache entry skips the rate limit and plugin. Otherwise, a bearer request with a delivery key claims its cache entry; all uncached requests pass the rate bucket.
6. Quivr opens the source credential and calls the plugin. The plugin verifies a signed POST over raw bytes, maps a bearer POST to items, or answers a GET challenge without items.
7. Quivr validates and ingests accepted POST items, records acceptance, then caches bearer answers or keeps successful signed reservations. Failed signed processing releases its reservation when cleanup succeeds.
8. Quivr commits the attempt’s audit and counters before replying with Receipts, a challenge answer or a refusal. Cached answers follow this audit step too.

​

Check it worked

​

Clean up

​

Next

• Receive and secure pushes for token lifecycle, limits, retries and audit.
• Write a connector for scheduled collection and attachments.
• Plugin protocol for the exact request and response fields.