Authoring Plugins¶

Imbi plugins extend the platform with integrations against external systems. Under Plugin Architecture v3 (api_version = 2) a Python package ships exactly one Plugin subclass whose manifest declares — once per package — the integration-level options and credentials plus a set of capabilities. Each capability binds one enumerated behavior (deployment, logs, identity, …) to an implementation class. The host services (imbi-api, and imbi-gateway for webhook actions) discover, validate, and instantiate plugins at runtime; plugins carry no global state and receive everything they need — context, credentials, and options — on every call.

An Integration graph node is a configuration instance of a plugin: it stores the resolved integration-level option values and the single encrypted credential blob. One plugin backs many Integrations. (The Integration node is the connection — there is no separate connection plugin type.)

This page covers the parts every plugin shares: discovery, the manifest, the Capability model, the request context, credential decryption, the registry, errors, templates, and plugin-declared graph schema. Each capability kind then has its own page documenting the contract base class it implements and the data models it exchanges with the host.

One Plugin per Package¶

A minimum plugin distribution contains:

1. One Plugin subclass whose class-level manifest: PluginManifest declares the package's slug, name, integration-level options and credentials, and its capabilities (each binding a handler class).
2. A module-level PLUGIN attribute in the package's root __init__, pointing at that Plugin subclass, so the convention scan can discover it.

# src/imbi/plugins/github/__init__.py
from imbi.plugins.github.plugin import GitHubPlugin

PLUGIN = GitHubPlugin

Discovery¶

load_plugins() finds plugins three ways, unions them, and dedupes by slug:

1. First-party scan — every subpackage of imbi.plugins (the plugins shipped with the imbi distribution) is imported and its module-level PLUGIN attribute (a Plugin subclass) is read. Plugin code always ships; a plugin whose optional dependencies are not installed (see the imbi[plugin-*] extras) is recorded as skipped, not as an error. Individual plugins can be turned off by slug with the IMBI_PLUGINS_DISABLED setting (comma-separated env var or list).
2. Convention scan — every installed top-level module named imbi_plugin_* is imported and its module-level PLUGIN attribute (a Plugin subclass) is read. This is the packaging convention for third-party plugins.
3. Explicit registration — the IMBI_PLUGINS setting (env var IMBI_PLUGINS, a comma-separated list — or JSON list — of dotted import paths such as mycorp.imbi.jira:JiraPlugin) covers packages that cannot follow the naming convention. It is read via imbi.common.settings.Plugins().imbi_plugins.

There is no entry-point group: the only source of truth is the class hierarchy, and discovery is a mechanical scan. Editable installs and monorepo dev work without re-running metadata hooks.

The Manifest¶

Every plugin declares a class-level manifest. Integration-level options and credentials are declared once; capabilities are enabled/configured per Integration. The host treats the manifest as immutable once registered.

from imbi.common.plugins import (
    Capability,
    CredentialField,
    Plugin,
    PluginManifest,
    PluginOption,
)

from imbi.plugins.github.capabilities import (
    GitHubDeployment,
    GitHubIdentity,
    GitHubLifecycle,
    GitHubWebhookActions,
    GitHubCommitSync,
    GitHubPullRequestSync,
)


class GitHubPlugin(Plugin):
    manifest = PluginManifest(
        slug='github',
        name='GitHub',
        description='Repositories, deployments, and identity on GitHub.',
        # Brand icon in `library-icon-name` form (Simple Icons, Tabler,
        # Lucide, …). Surfaced to the UI to show the provider's logo.
        icon='si-github',
        api_version=2,
        auth_type='oauth2',
        # Integration-level options — asked ONCE per Integration.
        options=[
            PluginOption(
                name='flavor',
                label='Flavor',
                choices=['github', 'ghec', 'ghes'],
                required=True,
            ),
            PluginOption(name='host', label='Host'),
        ],
        # Integration-level credentials — the ONLY credential declaration.
        credentials=[
            CredentialField(name='app_id', label='App ID', secret=False),
            CredentialField(name='private_key', label='Private key'),
            CredentialField(
                name='installation_id',
                label='Installation ID',
                secret=False,
            ),
            CredentialField(
                name='client_id',
                label='OAuth client ID',
                required=False,
                secret=False,
            ),
            CredentialField(
                name='client_secret',
                label='OAuth client secret',
                required=False,
            ),
        ],
        capabilities=[
            Capability(
                kind='identity',
                label='Sign in with GitHub',
                default_enabled=False,
                project_scoped=False,
                hints={'login_capable': True},
                handler=GitHubIdentity,
            ),
            Capability(
                kind='deployment',
                label='Deployments',
                hints={'supports_deployment_sync': True},
                handler=GitHubDeployment,
            ),
            Capability(
                kind='lifecycle',
                label='Repository lifecycle',
                hints={
                    'supports_lifecycle_sync': True,
                    'lifecycle_events': ['created', 'archived', 'deleted'],
                },
                handler=GitHubLifecycle,
            ),
            Capability(
                kind='webhook-actions',
                label='Webhook actions',
                handler=GitHubWebhookActions,
            ),
            Capability(
                kind='commit-sync',
                label='Commit history sync',
                handler=GitHubCommitSync,
            ),
            Capability(
                kind='pr-sync',
                label='Pull request sync',
                handler=GitHubPullRequestSync,
            ),
        ],
    )

Field notes¶

• slug is the canonical identifier. Two plugins with the same slug cannot coexist; the second is rejected at load.
• api_version must be 2. Plugins declaring other versions are skipped and reported as such, not errored.
• auth_type declares the credential flow: 'api_token' (default), 'oauth2', 'oidc', 'aws-iam-ic', or 'none'. OAuth client_id / client_secret are ordinary named fields in credentials.
• options are the integration-level option values collected once per Integration (e.g. GitHub flavor / host, AWS region). Five field types are supported: the four scalars (string, integer, boolean, secret) plus mapping (a key/value editor stored as dict[str, str]). Resolved values reach a capability via PluginContext.integration_options.
• credentials are the only credential declaration in the package — capabilities cannot declare their own. Values are encrypted at rest as the Integration's single blob and decrypted into the credentials dict passed to every call. Each field is secret=True by default (masked in the UI, never echoed back); set secret=False for non-sensitive identifiers such as an OAuth client_id or a GitHub App id so the UI renders them as plain text. Each field also accepts multiline, which defaults to False; set it to True for values that span multiple lines (e.g. a PEM private key) so the UI renders a textarea instead of a single-line input.
• capabilities must declare at least one Capability, and each kind must be unique within the manifest.
• data_types apply to configuration capabilities — see Configuration.
• vertex_labels / edge_labels let a plugin extend the AGE graph with its own reference data (see Plugin-declared Graph Schema).
• ops_log_templates supply Mustache-style templates the UI uses to render operations-log entries tagged with the plugin's slug, keyed by the action value the host wrote into the entry.

get_capability(kind: str) -> Capability | None

329
330
331
332
333
334

def get_capability(self, kind: str) -> Capability | None:
    """Return the capability of ``kind``, or ``None`` when absent."""
    for capability in self.capabilities:
        if capability.kind == kind:
            return capability
    return None

Capabilities¶

A Capability both declares operator-facing metadata (label, options, defaults) and binds the implementation via its handler — a class subclassing the contract base for the capability's kind. It is validated at construction: the handler must subclass the correct contract (CAPABILITY_CONTRACTS[kind]) and every hints key must be in the per-kind allowlist (HINT_ALLOWLIST).

Key fields:

• kind — one of the enumerated CapabilityKind values below.
• label / description — operator-facing text.
• options — capability-scoped PluginOptions, rendered under the capability's toggle in the Integration form; values reach a call via PluginContext.capability_options.
• default_enabled — initial toggle state when an Integration is created.
• project_scoped — whether the capability participates in per-project-type / per-project USES assignment. identity, for example, is Integration-wide, not project-scoped.
• requires_identity — the capability wants ctx.identity populated when available.
• hints — kind-specific hints, validated against HINT_ALLOWLIST (see table). These are the v1 manifest booleans, moved onto the capability.
• ui_module — RESERVED: package-relative path to a built ESM module the UI can load for the capability; None uses built-in UI.
• handler — the implementation class. Excluded from serialization; the manifest that reaches the API/UI is pure data.

Kinds, surfaces, and contracts¶

Each kind maps to exactly one contract ABC (CAPABILITY_CONTRACTS) and a fixed set of surfaces (CAPABILITY_SURFACES). Adding a kind is a deliberate base-model change.

Hint allowlist¶

cacheable is accepted for every kind (a hint the host may consult to cache a capability's reads). The remaining keys are per-kind:

An unknown hint key fails at manifest construction, so typos surface at load.

surfaces: frozenset[str]

Plugin Context¶

Every capability call receives a PluginContext carrying the resolved project's identity plus host-populated fields the capability can derive state from. A handful of fields are write-only side channels the capability sets and the host reads back after the call — most notably link_writeback and service_writeback.

Resolved configuration reaches a call on three layers:

• integration_options — the Integration's resolved integration-level option values (e.g. GitHub host/flavor, AWS region), so a capability can read connection-level config without re-declaring it.
• capability_options — the invoked capability's own option values (from Integration.capabilities[kind].options, layered with any USES-edge overrides). Empty when the capability declares none.
• assignment_options — the per-assignment (USES-edge) options.

The host also injects the running capability's bound Integration (integration_slug) and the project's EXISTS_IN connections (service_connections, each a ServiceConnection) so a capability can read the canonical project↔Integration relationship without re-querying the graph. A lifecycle capability maintains that relationship by setting service_writeback; the host persists it as the EXISTS_IN edge against the bound Integration.

resolve_user_by_identity is an optional host-injected coroutine that maps an external identity subject (e.g. a provider's numeric user id) to the matching Imbi user's email, or None when no active identity matches. A capability uses it to attribute external actors — such as the authors of synced commits — to Imbi users without knowing how the host reaches the identity store: the gateway wires an HTTP /users/by-identity lookup, the imbi-api worker a direct graph query. It is a live callable rather than data, so it is excluded from serialization and is None on any deserialized context. Callers should cache results.

Credentials¶

There is exactly one credential store per Integration — Integration.encrypted_credentials, a mapping of field name to a Fernet-encrypted value. Decrypt it with decrypt_integration_credentials, which accepts either the Integration node (anything with an encrypted_credentials mapping) or the mapping itself and returns the decrypted dict. There is no sibling lookup and no fallback ordering: every capability of an Integration receives the same decrypted blob (the credentials argument on every contract method).

from imbi.common.plugins import decrypt_integration_credentials

credentials = decrypt_integration_credentials(integration)

decrypt_integration_credentials(
    source: _HasEncryptedCredentials | dict[str, str],
) -> dict[str, str]

24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

def decrypt_integration_credentials(
    source: _HasEncryptedCredentials | dict[str, str],
) -> dict[str, str]:
    """Decrypt an Integration's credential blob.

    Accepts either an ``Integration`` node (anything with an
    ``encrypted_credentials`` mapping) or the mapping itself. Each value
    is Fernet-decrypted via :class:`TokenEncryption`; entries whose
    ciphertext is empty or fails to decrypt are dropped rather than
    surfaced as an empty string a ``key in dict`` check would satisfy.
    """
    if isinstance(source, dict):
        encrypted = source
    else:
        encrypted = source.encrypted_credentials

    encryptor = TokenEncryption.get_instance()
    decrypted: dict[str, str] = {}
    for name, ciphertext in encrypted.items():
        if not ciphertext:
            continue
        try:
            plaintext = encryptor.decrypt(ciphertext)
        except Exception:  # noqa: BLE001 - treat as missing, log and skip
            LOGGER.warning(
                'Integration credential %r failed to decrypt; skipping',
                name,
            )
            continue
        if plaintext:
            decrypted[name] = plaintext
    return decrypted

Search Templates¶

For capabilities that build provider-specific query strings from project context, use expand_template. It substitutes only the whitelisted variables — project_slug, org_slug, team_slug, environment, project_id, project_type_slug — written as ${name}, and rejects anything else:

from imbi.common.plugins import expand_template

label_query = expand_template(
    template,  # e.g. '{app="${project_slug}", env="${environment}"}'
    {
        'project_slug': ctx.project_slug,
        'org_slug': ctx.org_slug,
        'team_slug': ctx.team_slug,
        'environment': ctx.environment,
        'project_id': ctx.project_id,
        'project_type_slug': ctx.project_type_slugs[0],
    },
)

Validate templates at assignment time with validate_template so configuration errors surface early instead of during a search.

validate_template(template: str) -> None

19
20
21
22
23
24
25
26
27
28
29
30
31

def validate_template(template: str) -> None:
    """Validate a search template string, rejecting unknown variables.

    Raises:
        ValueError: If the template references variables not in the whitelist.
    """
    for match in _VAR_PATTERN.finditer(template):
        var = match.group(1)
        if var not in _ALLOWED_VARS:
            raise ValueError(
                f'Unknown template variable ${{{var}}};'
                f' allowed: {sorted(_ALLOWED_VARS)}'
            )

expand_template(
    template: str, variables: dict[str, str | None]
) -> str

34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

def expand_template(
    template: str,
    variables: dict[str, str | None],
) -> str:
    """Expand a search template substituting whitelisted variables.

    Absent variable values are replaced with empty strings. The template is
    validated first to reject unknown placeholders, preserving the whitelist
    guarantee from :func:`validate_template`.
    """
    validate_template(template)

    def _sub(match: re.Match[str]) -> str:
        var = match.group(1)
        val = variables.get(var)
        return val if val is not None else ''

    return _VAR_PATTERN.sub(_sub, template)

Registry and Loading¶

The host calls load_plugins() at startup (and reload_plugins() on demand) to populate the registry from the convention scan unioned with IMBI_PLUGINS. Every discovered plugin is validated fail-loud at load: the class must subclass Plugin; its manifest must be a PluginManifest with a supported api_version; every capability's handler must subclass the contract for its kind; webhook-actions and tools catalogs must enumerate cleanly with unique names; declared vertex/edge labels must not collide; and plugin slugs must be unique.

Lookups are by plugin slug — get_plugin(slug) returns a RegistryEntry (plugin_cls, manifest, package_name, package_version) — and by (slug, kind) — get_capability(slug, kind) returns the handler class for that capability. list_plugins() returns every registered entry. load_plugins() / reload_plugins() return a LoadResult (loaded, errors, skipped).

load_plugins() -> LoadResult

269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368

def load_plugins() -> LoadResult:
    """Discover, validate, and load all installed imbi plugins."""
    from imbi.common import settings

    plugin_settings = settings.Plugins()
    disabled = set(plugin_settings.imbi_plugins_disabled)

    loaded: list[str] = []
    errors: dict[str, str] = {}
    skipped: list[str] = []
    new_registry: dict[str, RegistryEntry] = {}
    seen_ids: set[int] = set()

    def _register(source: str, cls: object, version: str) -> None:
        if id(cls) in seen_ids:
            return
        entry, error, was_skipped = _validate_plugin(source, cls, version)
        if was_skipped:
            skipped.append(source)
            return
        if error is not None:
            LOGGER.error('Plugin load failed: %s', error)
            errors[source] = error
            return
        if entry is None:  # unreachable: entry set when no error / skip
            return
        if entry.manifest.slug in disabled:
            LOGGER.info(
                'Plugin %r disabled via IMBI_PLUGINS_DISABLED; skipping',
                entry.manifest.slug,
            )
            skipped.append(source)
            return
        seen_ids.add(id(cls))
        if entry.manifest.slug in new_registry:
            msg = f'{source}: duplicate plugin slug {entry.manifest.slug!r}'
            LOGGER.error(msg)
            errors[source] = msg
            return
        new_registry[entry.manifest.slug] = entry
        loaded.append(entry.manifest.slug)
        LOGGER.info(
            'Loaded plugin %r v%s (slug=%r, capabilities=%s)',
            entry.package_name,
            entry.package_version,
            entry.manifest.slug,
            [c.kind for c in entry.manifest.capabilities],
        )

    for module_name, source in _discover_first_party().items():
        try:
            cls = _load_convention_plugin(module_name)
        except ImportError as exc:
            # First-party plugin code is workspace-editable in dev; a
            # missing optional dependency means the plugin's
            # distribution is not installed, which is a normal
            # deployment state.
            LOGGER.debug(
                'Skipping first-party plugin %r: %s (install the matching '
                'imbi-plugin-* distribution to enable it)',
                module_name,
                exc,
            )
            skipped.append(source)
            continue
        except Exception as exc:
            LOGGER.exception('Failed to import plugin %r', module_name)
            errors[source] = str(exc)
            continue
        _register(source, cls, _first_party_version(module_name))

    for module_name, source in _discover_convention().items():
        try:
            cls = _load_convention_plugin(module_name)
        except Exception as exc:
            LOGGER.exception('Failed to import plugin %r', module_name)
            errors[source] = str(exc)
            continue
        _register(source, cls, _package_metadata(module_name)[1])

    for dotted_path in plugin_settings.imbi_plugins:
        try:
            module_name, cls = _load_explicit_plugin(dotted_path)
        except Exception as exc:
            LOGGER.exception('Failed to import IMBI_PLUGINS %r', dotted_path)
            errors[dotted_path] = str(exc)
            continue
        _register(dotted_path, cls, _package_metadata(module_name)[1])

    # Refuse vlabel/edge collisions across loaded plugins or with core
    # schemata.  Imported lazily to avoid a circular import.
    from imbi.common.plugins.schemas import validate_no_collisions

    validate_no_collisions([entry.manifest for entry in new_registry.values()])

    with _lock:
        _registry.clear()
        _registry.update(new_registry)

    return LoadResult(loaded=loaded, errors=errors, skipped=skipped)

reload_plugins() -> LoadResult

371
372
373
374

def reload_plugins() -> LoadResult:
    """Reload the plugin registry from installed packages."""
    LOGGER.info('Reloading plugin registry')
    return load_plugins()

get_plugin(slug: str) -> RegistryEntry

377
378
379
380
381
382
383
384
385
386
387

def get_plugin(slug: str) -> RegistryEntry:
    """Get a registry entry by plugin slug.

    Raises:
        PluginNotFoundError: If the slug is not registered.
    """
    with _lock:
        entry = _registry.get(slug)
    if entry is None:
        raise PluginNotFoundError(slug)
    return entry

get_capability(
    slug: str, kind: str
) -> type[CapabilityHandler]

390
391
392
393
394
395
396
397
398
399
400
401

def get_capability(slug: str, kind: str) -> type[CapabilityHandler]:
    """Return the handler class for ``kind`` on plugin ``slug``.

    Raises:
        PluginNotFoundError: If the slug is not registered or the plugin
            declares no capability of ``kind``.
    """
    entry = get_plugin(slug)
    capability = entry.manifest.get_capability(kind)
    if capability is None:
        raise PluginNotFoundError(f'{slug}:{kind}')
    return typing.cast(type[CapabilityHandler], capability.handler)

list_plugins() -> list[RegistryEntry]

404
405
406
407

def list_plugins() -> list[RegistryEntry]:
    """Return all registered plugins."""
    with _lock:
        return list(_registry.values())

RegistryEntry(
    plugin_cls: type[Plugin],
    manifest: PluginManifest,
    package_name: str,
    package_version: str,
)

Plugin-declared Graph Schema¶

A plugin may extend the AGE graph with its own vertex labels and edges via PluginManifest.vertex_labels / edge_labels. The host refuses any declaration that collides with core schemata or with another plugin's differently-shaped label (validate_no_collisions) and creates the declared vlabels and indexes idempotently on startup (apply_plugin_schemas).

validate_no_collisions(
    manifests: list[PluginManifest],
    schemata_toml_path: Path | None = None,
) -> None

 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

def validate_no_collisions(
    manifests: list[PluginManifest],
    schemata_toml_path: pathlib.Path | None = None,
) -> None:
    """Refuse vlabel/elabel name collisions across plugins / core.

    Logs at ERROR and raises :class:`PluginSchemaCollisionError` on the
    first collision detected.
    """
    core_names = _load_core_vlabel_names(schemata_toml_path)
    seen_vlabels: dict[str, tuple[str, PluginVertexLabel]] = {}
    seen_edges: dict[str, tuple[str, PluginEdgeLabel]] = {}

    for manifest in manifests:
        for vlabel in manifest.vertex_labels:
            if vlabel.name in core_names:
                LOGGER.error(
                    'Plugin %r declares vlabel %r which collides with '
                    'core schemata',
                    manifest.slug,
                    vlabel.name,
                )
                raise PluginSchemaCollisionError(
                    f'Plugin {manifest.slug!r} declares vlabel '
                    f'{vlabel.name!r} which collides with core schemata'
                )
            existing = seen_vlabels.get(vlabel.name)
            if existing is not None:
                owner_slug, owner_vlabel = existing
                if owner_vlabel != vlabel:
                    LOGGER.error(
                        'Plugin %r declares vlabel %r with a shape that '
                        'differs from plugin %r',
                        manifest.slug,
                        vlabel.name,
                        owner_slug,
                    )
                    raise PluginSchemaCollisionError(
                        f'Plugin {manifest.slug!r} declares vlabel '
                        f'{vlabel.name!r} with a shape that differs '
                        f'from plugin {owner_slug!r}'
                    )
                continue
            seen_vlabels[vlabel.name] = (manifest.slug, vlabel)

        for elabel in manifest.edge_labels:
            existing_edge = seen_edges.get(elabel.name)
            if existing_edge is not None:
                owner_slug, owner_edge = existing_edge
                if owner_edge != elabel:
                    LOGGER.error(
                        'Plugin %r declares edge label %r with a shape '
                        'that differs from plugin %r',
                        manifest.slug,
                        elabel.name,
                        owner_slug,
                    )
                    raise PluginSchemaCollisionError(
                        f'Plugin {manifest.slug!r} declares edge label '
                        f'{elabel.name!r} with a shape that differs '
                        f'from plugin {owner_slug!r}'
                    )
                continue
            seen_edges[elabel.name] = (manifest.slug, elabel)

apply_plugin_schemas(
    manifests: list[PluginManifest],
) -> None

118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

async def apply_plugin_schemas(
    manifests: list[PluginManifest],
) -> None:
    """Apply each plugin's declared vlabels + indexes to AGE.

    Mirrors :func:`imbi.common.graph.initializer.initialize` primitives.
    Idempotent — safe to invoke on every startup.
    """
    if not manifests:
        return

    validate_no_collisions(manifests)

    postgres = settings.Postgres()

    async with await psycopg.AsyncConnection.connect(
        str(postgres.url),
        autocommit=True,
    ) as conn:
        await conn.execute(
            'SET search_path = ag_catalog, "$user", public',
        )
        async with conn.cursor() as cursor:
            for manifest in manifests:
                for vlabel in manifest.vertex_labels:
                    await _ensure_vlabel(
                        cursor, postgres.graph_name, vlabel.name
                    )
                    for index in vlabel.indexes:
                        await _ensure_vlabel_index(
                            cursor,
                            postgres.graph_name,
                            vlabel.name,
                            index.fields,
                            index.unique,
                        )

Errors¶

Capabilities should raise exceptions from imbi.common.plugins.errors when the failure mode maps onto one of them; otherwise let the host wrap the exception. The author-relevant ones:

PluginNotFoundError is host-side (registry lookups) — capability code should not raise it.

PluginRateLimited(retry_at: float, message: str = '')

49
50
51

def __init__(self, retry_at: float, message: str = '') -> None:
    self.retry_at: float = retry_at
    super().__init__(message or f'Rate limited until epoch {retry_at:.0f}')

PluginRemediationNotSupported(
    plugin_slug: str, remediation_id: str
)

62
63
64
65
66
67
68

def __init__(self, plugin_slug: str, remediation_id: str) -> None:
    self.plugin_slug: str = plugin_slug
    self.remediation_id: str = remediation_id
    super().__init__(
        f'Plugin {plugin_slug!r} does not support remediation '
        f'(id={remediation_id!r})'
    )

Lifecycle and State¶

A new handler instance is created per request. Do not stash connection state, cached credentials, or per-project data on self. If you need a connection pool or rate-limited HTTP client, construct it inside the method, scoped to the call:

async def list_keys(self, ctx, credentials):
    async with httpx.AsyncClient(timeout=10.0) as client:
        ...

Testing¶

Capabilities can be unit-tested in isolation: instantiate the handler class directly and call its methods with synthesized PluginContext objects and a credentials dict. For registry-level integration tests, see tests/test_plugins/test_registry.py in this repository for examples of injecting a plugin without installing a real distribution.