dagstack-plugin-system 0.4.1

Plugin-system core for dagstack ecosystem — pluggy-based, PEP 420 namespace under dagstack.plugin_system, MCP adapters

dagstack/plugin-system-python

Orchestration-neutral plugin framework — Python implementation. Implements dagstack/plugin-system-spec on top of pluggy + pydantic. Sister implementations: @dagstack/plugin-system (TypeScript) and go.dagstack.dev/plugin-system (Go). A pluggable extension system where plugins behave the same in a FastAPI process, in Dagster ops, in a Celery task, or in a pytest fixture — without knowing the host.

Status: release-candidate (Phase 1 MVP, 0.1.0). The first public PyPI release follows after API stabilisation and a soak period.

---

What it solves

The Python ecosystem provides pluggy (discovery + hook invocation) and several orchestrators (Dagster, Airflow, Celery, FastAPI background tasks), but the contract between plugin and host is fuzzy: a plugin written for one runtime often does not survive in another without rework.

dagstack/plugin-system-python formalises that contract via eight runtime invariants (ambient-state ban, serialisable boundaries, DI-injected resources, sync/async declaration, partition keys, abstract progress/checkpoint sinks, content-hash idempotency, determinism boundary) and five dispatch classes (Singleton / Broadcast-Collect / Broadcast-Notify / Chain / Dispatch-by-capability) on top of pluggy.

Result:

• the same plugin runs in FastAPI in-process, in a Dagster dynamic-partitioned asset, in a Celery task, in a test — without changes;
• idempotent incremental work out of the box — the orchestrator skips a UoW whose content_key is already in storage;
• 3 runtimes — in_process (Python), mcp_stdio (subprocess in any language), mcp_http (remote service) — chosen per plugin;
• contract-test framework — mandatory checks for plugin authors (round-trip serialisation, lifecycle leak detection, ambient-state sniffing, determinism AST check).

Who it's for

• Teams building extensible Python applications with several integration points (data sources, pre/post processors, backends) and wanting every point to be pluggable.
• Authors of RAG / agent platforms who need to swap LLM backends, vector stores, content-source adapters without changing the core.
• Teams who need to scale a current in-process monolithic architecture onto an orchestrator (Dagster, Celery, k8s) without rewriting plugins.

Position relative to existing tools

Tool	What it provides	What it doesn't cover (and dagstack does)
pluggy	discovery, hook invocation, lifecycle	runtime invariants, out-of-process adapters, contract tests
stevedore	discovery via entry_points	everything beyond discovery
Dagster core	orchestration, UoW, idempotency	discovery/hooks for IDE-embedded extensions, MCP runtime
Airflow providers	pluggable operators/hooks	runtime-neutrality, a 3rd runtime over HTTP

dagstack does not replace Dagster/Airflow — it provides the contract that lets the same plugin run inside any of those orchestrators and inside a plain web service at the same time.

Quickstart

End-to-end example: registering the built-in echo plugin (examples/echo_plugin/) and running its lifecycle through the public API. The same code path is exercised by the e2e test tests/e2e/test_echo_plugin.py.

import asyncio
import logging

from dagstack import PluginContext, PluginRegistry
from examples import echo_plugin


async def main() -> None:
    registry = PluginRegistry()
    registry.register_module(echo_plugin)

    ctx = PluginContext(
        config={},
        logger=logging.getLogger("demo"),
        registry=registry,
    )
    await registry.setup_all(ctx)

    plugin = registry.get_plugin("tool", name="echo")
    print(plugin.execute({"msg": "hi"}))  # → {"echoed": "hi"}

    # And the same call through the pluggy hook (Singleton, firstresult=True):
    print(registry.plugin_manager.hook.execute(args={"msg": "via-pluggy"}))

    await registry.teardown_all()


asyncio.run(main())

The plugin manifest in examples/echo_plugin/dagstack_plugin.toml:

[tool.dagstack.plugin]
schema_version = "1"
name = "echo"
kind = "tool"
kind_api_version = "1"
dagstack_version = ">=0.1.0,<1.0.0"
runtime = "in_process"
license = "Apache-2.0"
entry_point = "examples.echo_plugin.echo_plugin:EchoTool"

See also examples/echo_plugin/README.md — what the example demonstrates and which part of dagstack it covers as a smoke test.

Writing plugins — common pitfalls

Don't use @property for plugin-state fields

pluggy.PluginManager.register(instance, name=...) calls inspect.getmembers(instance) to discover @hookimpl-decorated methods. That triggers every property getter on the instance — including those that declare "not ready until setup()". If a property raises RuntimeError("accessed before setup"), discover() catches it under the continue-on-failure policy and silently skips the plugin.

Bad:

class MyPlugin:
    def __init__(self):
        self._client = None

    @property
    def client(self):          # ← pluggy triggers this before setup()
        if self._client is None:
            raise RuntimeError("accessed before setup()")
        return self._client

Good — a plain attribute, None until setup:

class MyPlugin:
    def __init__(self):
        self.client = None     # plain attribute, pluggy-safe

    async def setup(self, ctx):
        self.client = make_client()

    async def teardown(self):
        self.client = None

The same advice applies to any computed attribute that depends on setup() state. If the public contract requires lazy initialisation, expose it explicitly as a method (get_client()) instead of a property.

Thread-safety

PluginRegistry is not thread-safe. The host MUST serialise calls to discover(), register_*(), setup_all(), teardown_all() — in the standard case (FastAPI lifespan / asyncio single-loop) this is already guaranteed. Parallel calls from different threads race on _loaded and _setup_done; in that case wrap access in an external lock or use one registry per loop.

Pilot integration

A pilot integration drives the design — a RAG-style code-search platform refactoring its core onto a plugin architecture, with kinds such as LLMClient, Chunker, VectorStore, RAGPipeline, AgentTool, Embedder, BlobSource, VcsSource, DocumentSource, ContentRenderer, Orchestrator implemented on top of the dagstack contract. The pilot validates the runtime invariants and dispatch classes against a real production workload before the open-source release.

Architecture

The architectural decisions are recorded in the dagstack ADR series:

• docs/adr/0001-plugin-architecture-core.md
• docs/adr/0002-hook-invocation-semantics.md
• docs/adr/0003-orchestration-neutral-runtime.md
• docs/adr/README.md — index, numbering convention, provenance

The full normative ADRs (language-agnostic) live in dagstack/plugin-system-spec.

In short:

• Plugin Manifest — a pydantic model (dagstack_plugin.toml / [tool.dagstack.plugin] in pyproject.toml) with fields: name, kind, runtime, core_version, capabilities, supports_* for dispatch, execution_model, unit_of_work, resources.required.
• PluginRegistry — a thin layer over pluggy.PluginManager: version gates, lifecycle, runtime adapters.
• 3 runtime adapters — InProcessAdapter (Python class), MCPStdioAdapter (subprocess + JSON-RPC over stdin/stdout), MCPHttpAdapter (SSE / Streamable HTTP).
• Resources as an open registry — ctx.resources.get("http_client" | "clock" | "rng" | "postgres" | ...); required/optional declared in the manifest, statically gated by the registry.
• Progress + Checkpoint sinks — abstract, host-provided; the plugin does not know where progress lands (WebSocket / Dagster AssetMaterialization / InMemory).
• Contract-test framework — assert_lifecycle_clean, assert_orchestration_neutral, assert_json_serializable_boundaries, assert_resources_via_di, assert_deterministic (for output_hash-idempotent plugins).

Roadmap

Phase 0 — Foundation — done.

• Lift the plugin-architecture ADRs into the dedicated dagstack series docs/adr/0001..0003.
• Bootstrap pyproject.toml, pytest config, pre-commit, CI.
• Core API skeleton: PluginManifest, PluginRegistry, PluginContext, InProcessAdapter, baseline hookspecs, contract-test stubs.
• Built-in echo plugin as a smoke test.

Phase 1 — MVP API — done (0.1.0).

• 5 dispatch classes: Singleton, Broadcast-Collect, Broadcast-Notify, Chain, Capability.
• Lifecycle with topo-sort (depends_on) and continue-on-failure.
• Resources DI runtime + metadata propagation for governance patterns.
• ProgressSink + CheckpointStore reference implementations.
• Governance filter callback for capability dispatch.
• 258 tests, ~94% coverage.

Phase 2 — Production readiness.

• MCP stdio / HTTP adapters (ADR-0001 §Runtime adapters).
• Contract-test framework v1 (full invariant suite).
• Pilot consumer integration.
• Public docs site (in flight at plugin-system.dagstack.dev).

Phase 3 — Open source release.

• API stabilisation, semver 1.0.0.
• PyPI public release.
• GitHub mirror flips public.

License

Apache License 2.0.

Contact

Repository maintainer: Evgenii Demchenko, [e-mail].