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

# Quivr Plugin SDK for Python

> Write, test and run a Python normalizer

`quivr-plugin-sdk` (import `quivr_plugin`) implements the Plugin Protocol v0
([contract](/contracts/plugins/v0), Plugin API 0.2) so that a
Python normalizer or alert rule is a single function. It has no Temporal, Weaviate or database clients, and its
only runtime dependencies are PyYAML and jsonschema (both MIT). Python 3.12 or
later.

The SDK is installed from this repository; it is not published to PyPI in v0.

```bash theme={null}
pip install -e sdks/python                       # from a checkout
pip install "quivr-plugin-sdk @ git+https://github.com/The-Vibe-Company/quivr-v2#subdirectory=sdks/python"
```

Start a new plugin with `quivr plugin init <name>`. It writes a working
`text/markdown` normalizer, a fixture and tests that use only this SDK.
`quivr plugin init <name> --kind subscription` writes an alert rule instead.

<h2 id="a-normalizer">
  A normalizer
</h2>

```python theme={null}
from pathlib import Path
from quivr_plugin import Invocation, ManifestContent, NormalizerResponse, Part, Plugin, TerminalError, TextContent

plugin = Plugin(Path(__file__).parent.parent / "quivr-plugin.yaml")

@plugin.normalizer
def normalize(invocation: Invocation) -> NormalizerResponse:
    text = invocation.read_input().decode("utf-8")   # verified against size and sha256
    if not text.strip():
        raise TerminalError("empty_document", "the document has no text")
    return NormalizerResponse(manifest=ManifestContent(parts=[
        Part(key="body", role="body", content=TextContent(text=text)),
    ]))

if __name__ == "__main__":
    plugin.serve()   # QUIVR_PLUGIN_HOST / QUIVR_PLUGIN_PORT, else 127.0.0.1:8080
```

<h2 id="an-alert-rule">
  An alert rule
</h2>

An alert rule is the `subscription` Contribution. It receives one Record
Version's text Parts and a batch of distinct evaluations (a Saved Query
`expression` and a Subscription `configuration` each, already validated against
the manifest's `expression_schema` and `configuration_schema`), and returns one
decision per evaluation:

```python theme={null}
from quivr_plugin import Plugin, SubscriptionInvocation, match, no_match

plugin = Plugin(Path(__file__).parent.parent / "quivr-plugin.yaml")

@plugin.subscription
def evaluate(invocation: SubscriptionInvocation):
    decisions = []
    for evaluation in invocation.evaluations:
        needle = evaluation.expression["text"].casefold()
        keys = [p.key for p in invocation.parts if needle in p.text.casefold()]
        if keys:
            decisions.append(match(evaluation, "The phrase appears.", part_keys=keys))
        else:
            decisions.append(no_match(evaluation))
    return decisions
```

`not_ready(evaluation)` defers a decision, for example until
`invocation.enriched`. A decision must depend only on the record, the
expression and the configurations: the core batches, deduplicates and replays
evaluations freely.

<h2 id="what-the-sdk-does">
  What the SDK does
</h2>

| Concern | Behavior |
| - | - |
| Models | Dataclasses generated from the contract schemas (`quivr_plugin.models`), with `from_dict` and `to_dict` |
| Routes | `GET /v0/discovery` (plugin identity, the declared Contributions, the highest Plugin API version the manifest range admits — `0.1.0` or `0.2.0` —, `sha256:` digest of the exact `quivr-plugin.yaml` bytes), `GET /v0/health`, `POST /v0/contributions/normalizer`, `POST /v0/contributions/subscription` |
| Request checks | Request schema → 400 `invalid_request`; media type not declared → 400 `unsupported_media_type`; configuration against the manifest configuration schema → 400 `invalid_configuration`; a subscription expression or evaluation configuration against the declared schemas → 400 `invalid_expression` or `invalid_subscription_configuration` |
| Errors | `RetryableError` → 503, `retryable: true`. `TerminalError` → 422, `retryable: false`. Unexpected exception → 500 `internal_error`, `retryable: false`. Always the protocol error envelope |
| Response checks | Before sending: response schema (500 `invalid_response`) and the declared `max_response_bytes` (500 `response_too_large`). The engine still applies its own Manifest validation. For a subscription, also one decision per evaluation, evidence for every match, Part keys that exist and details of at most 16 KiB (500 `invalid_response`) |
| Input Blob | `Invocation.read_input()` reads `file://` or signed http(s) references, at most `size_bytes + 1` bytes. Transport errors and 401/403/408/425/429/5xx raise `RetryableError("input_unavailable")`; a size or SHA-256 mismatch raises `TerminalError` |
| Logging | `serve()` logs JSON lines to stderr. Every record emitted during an invocation carries `invocation_id` and `idempotency_key`; use `invocation.logger` or any logger |
| Health | `@plugin.health_check` may raise a `PluginError` to answer 503 while not ready |

<h2 id="testing-a-plugin">
  Testing a plugin
</h2>

```python theme={null}
from quivr_plugin.testing import expect_response, invoke_fixture

response = expect_response(invoke_fixture(plugin, "fixtures/sample.json"))
```

`invoke_fixture` builds the same request `quivr plugin dev --fixture` sends,
using an invocation fixture (`contracts/plugins/v0/plugin-fixture.schema.json`),
and runs the normalizer route in process. `Plugin.invoke(request)` and
`Plugin.handle(method, path, body)` expose the same dispatch without HTTP.

For an alert rule, `invoke_subscription_fixture(plugin, "fixtures/sample.json")`
builds the same batches as `quivr plugin dev` and the Contract Runner from a
subscription fixture (`contracts/plugins/v0/subscription-fixture.schema.json`),
runs them through `Plugin.evaluate(request)`, and fails when a decision
differs from the fixture's `expect`. `build_subscription_requests` returns the
requests themselves.

<h2 id="maintaining-the-sdk">
  Maintaining the SDK
</h2>

```bash theme={null}
python3 sdks/python/scripts/generate.py          # regenerate models.py and schema copies after a contract change
python3 sdks/python/scripts/generate.py --check  # fails when they are stale (part of make test)
bash scripts/plugin_sdk.sh                       # SDK tests and a scaffolded plugin end to end (part of make test)
```

The generator is standard-library only, so the output is reproducible. It
reads `contracts/shared/v0/manifest.schema.json` and
`contracts/plugins/v0/*.schema.json`. Never edit `models.py` or
`src/quivr_plugin/schemas/` by hand.