Plugins

Everything dirigent discovers and contributes is a plugin: blocks, connection kinds, storage backends, notifiers, and the format checkers a schema asserts against. There is one framework for all of it -- pluginkit -- and one seam between a pack and a running instance: a Python entry point. A pack lives in its own repository, is installed into an instance's environment, and is wired in by that entry point alone. Install is the whole of configuration.

dirigent-plugin is the contract a pack builds against. The worked example throughout this page is dirigent-acme, a pack for an imaginary vendor, Acme: it contributes an acme connection kind, the acme.* blocks that speak to it, and the acme-site-id format.

The seam is one entry point

A pack declares a single entry point under the dirigent.plugins.v1 group, pointing at its plugin object:

# the pack's pyproject.toml
[project.entry-points."dirigent.plugins.v1"]
acme = "dirigent_acme:plugin"

That object exposes one @extension-marked method returning a Contribution:

from dirigent_plugin import Contribution, extension


class AcmePlugin:
    @extension
    def contribute(self) -> Contribution:
        return Contribution(
            operators=[AcmeSitesOperator(), AcmeOrdersOperator(), ...],
            sensors=[AcmeOrderShippedSensor()],
            connection_kinds=[AcmeConnectionKind()],
            formats={"acme-site-id": is_site_id},
            labels=[ACME],
        )


plugin = AcmePlugin()

At startup the host (dirigent_core.plugins.load_plugin_host, built on pluginkit's PluginManager) scans the dirigent.plugins.v1 group, finds every installed distribution that declares one, calls each contribute(), and merges the results into one catalog. The scan happens once; after that dispatch is direct. Two packs claiming the same block id is a startup error, not last-one-wins -- a block id is public API that stored pipelines reference by string forever.

So adding a pack to any instance is:

uv add dirigent-acme

into the same environment, then a restart. It appears in dg blocks and as a connection kind with no change to dirigent, no config file, and no registry to edit. Uninstall and it is gone.

What a pack contributes

contribute() returns one Contribution spanning the surfaces the host knows:

• operators -- the blocks that do work and hand a typed output on, a transform engine among them
• sensors -- blocks that hold a run open until a condition is met
• connection_kinds -- how an external system is described and authenticated, with a health check. HealthReport.healthy has three values, not two: True, False, and None for a check that ran and could not verify anything, which is what a kind answers rather than claiming either outcome
• sources -- a kind of source a pipeline's first node may be, keyed by the kind a document writes under source. A contributed kind is a PolledSource, a HeldSource or a ReceivedSource. A HeldSource answers listen(config, ctx) with an async context manager around a consumer whose receive(limit, wait) hands over inputs and whose ack() acknowledges every input received so far; the worker calls ack() only once those inputs are runs, and ctx.connection(ref, model) resolves the connection the consumer reaches its broker through. A PolledSource answers poll(config, cursor, ctx, limit) with the inputs it found, oldest first, and the cursor the next poll carries on from; a worker calls it every poll_every(config), with the instance's storage on ctx, and a poll that raises is recorded on the source and backed off from the same cursor. For a ReceivedSource the server answers its route under /endpoints, checks the head with admit, and hands receive the read request, which answers with an input, Ignore for a sign of life such as a handshake (answered 204, no run), or a Refusal carrying the status its caller gets. receive runs in the server process with no connection and no secret in reach. endpoint is dirigent's own kind, and a pack cannot take its name. dirigent_testing.receive_request takes a request through a kind the way the intake does.
• storage_backends -- a scheme a <scheme>:// URI writes bytes to
• notifiers -- where an alert is sent
• formats -- a JSON Schema format checker by name, so a schema that writes format: <name> asserts wherever the pack is installed and stays a passing annotation where it is not
• labels -- the pack's own Catalogues, so a surface can render the pack's refusals in its own language. A pack's refusal reaches a browser under the pack's own code, and the browser holds no table for it; contributing the catalogue is what puts one there. The instance serves every contributed code and its template from GET /labels, and the web UI renders a refusal from its code rather than repeating the English the worker happened to mint. One prefix has one owner, so two packs contributing the same prefix is a startup error like a block id

A Contribution carries an api_version: a pack written against a different revision of the contract is refused at load, cleanly, rather than half-working. The host merges every pack's contribution and rejects id collisions across all surfaces.

A pack writes RST, and the catalog serves markdown

A block's user-facing documentation is its field docstrings, which are Python docstrings and so are written in reStructuredText. Every reader of the catalog -- the UI's block panel and step form, dg blocks show, and the generated block reference -- renders markdown. dirigent_common.docstrings.as_markdown is where the two meet, applied once at the catalog boundary: dirigent_core.plugins.json_schema walks the schema a contributed model publishes and rewrites every description on the way out.

class CheckoutConfig(BlockModel):
    ref: str = "main"
    """The branch, tag or commit to check out. ``HEAD`` is not a ref."""

The catalog serves that description with HEAD in single backticks. So a pack author writes RST inline literals -- double backticks -- and gets markdown inline code everywhere. That rewrite is the whole of the translation: inline literals and nothing else, so RST that has no markdown equivalent reaches a reader unchanged. A description is one or two plain sentences with code in double backticks.

A config field that names another thing says so

A block whose config takes the code of a connection or of a schema types that field ConnectionRef or SchemaRef rather than str:

class CheckoutConfig(BlockModel):
    connection: ConnectionRef
    """The ``git`` connection naming the remote and holding its credential."""

Both are str to every validator, and both publish x-dirigent-ref in the schema the catalog serves. What reads it is the step form: a field typed either way is drawn with the thing it names under it -- the connection's settings and health, or the schema's body -- and a link to that thing's own screen.

A connection field says which kinds it takes

A field that reads a connection through its own model takes one kind, and says which, so a step naming a connection of another kind is refused at plan and apply rather than failing mid-run:

type GitConnectionRef = Annotated[str, Reference("connection", kinds=("git",))]


class CheckoutConfig(BlockModel):
    connection: GitConnectionRef
    """The ``git`` connection naming the remote and holding its credential."""

A field that reads a connection by its fields, as ctx.http does, says which fields instead, and then takes a connection of any kind that carries them. The built-in HTTP blocks take an http connection or one of any kind carrying a base_url, which is how a pack's own HTTP-backed kind fits http.request with nothing to register:

type HttpConnectionRef = Annotated[str, Reference("connection", kinds=("http",), carrying=("base_url",))]

Either publishes x-dirigent-accepts beside x-dirigent-ref. The kind checked is the one the instance holds the connection under, or, for dg validate with no server, the one a document that carries its own connection declares; a code written as a ${...} reference is left to the run. A refusal is document.step_connection_wrong_kind, or document.step_connection_for_alerts when the connection's kind is one an alert channel delivers through, such as a webhook connection named by webhook.post. A plain ConnectionRef takes every kind.

A refusal a block makes carries a code

A block never refuses with a free string. Every refusal it can make is a catalogued Message: a stable dotted code, and one English template with named params. A pack keeps them in a messages.py of its own, under a Catalogue named for the pack:

from dirigent_common import Catalogue

ACME = Catalogue("acme")

NO_CREDENTIAL = ACME.define(
    "no_credential",
    "connection {connection} has no api_key, and {url} accepts none without one",
)

REJECTED = ACME.define("rejected", "{method} {url} answered {status}: {detail}")

BlockFailure takes the message and the params it renders, beside the classification the engine retries on:

raise BlockFailure(
    REJECTED,
    error_class=ErrorClass.REJECTED,
    method=config.method,
    url=request_url(config),
    status=response.status_code,
    detail=body.get("message", ""),
)

check_config answers with issues built the same way, from the same catalogue:

def check_config(self, config: BaseModel) -> list[Issue]:
    if not isinstance(config, AcmePushConfig) or config.batch_size <= MAX_BATCH:
        return []
    return [Issue.of(BATCH_TOO_LARGE, asked=config.batch_size, maximum=MAX_BATCH)]

A limit is cheapest on the field itself, where plan and validate see it before anything runs. Field(le=...) on an integer becomes the schema's maximum, and on a Duration or a Size, which JSON Schema has no bound for, it is published as x-dirigent-bounds -- {"kind": "duration", "maximum": "30s"} -- and enforced by every check that holds the catalog, dg validate offline and an MCP validate_pipeline included. A literal config is also validated against the block's model at apply, so a field or model validator refuses there too; an error on a value that is still a ${...} is left to the run. check_config is for what neither can state, and the run is for what depends on the data.

The attempt row, the run event and the problem document all carry the code and the params beside the rendered sentence, so an operator can select one refusal out of a stream with jq 'select(.error_code == "acme.rejected")' however the English is later reworded. A pack's prefix is its pack name, and it owns everything under it. A param never carries a secret: the connection's code names it, its credential does not appear.

A config's own validators are held to the same rule, and they are the one place it is easy to forget: pydantic owns the code a validator's refusal reaches the wire under, so the message is rendered into the ValueError it wraps rather than given to it.

@model_validator(mode="after")
def _needs_a_credential(self) -> "AcmeConnectionConfig":
    if self.api_key is None:
        raise ValueError(NO_CREDENTIAL.render(connection=self.code, url=self.base_url))
    return self

Checking that every refusal carries one

dirigent-testing ships the check, so a pack holds its own wording to this page without reaching into the engine's repository. It is one call, and nothing is autouse:

from pathlib import Path

import dirigent_acme
from dirigent_testing import check_pack_messages


def test_every_refusal_the_pack_makes_carries_a_code():
    assert check_pack_messages(Path(dirigent_acme.__file__).parent) == []

It reads every module under the directory and reports three things: a refusal raised under no code, a message minted and never read, and a code two catalogues both define. That last walk is over every catalogue the process has imported -- the engine's and the pack's -- so a prefix a pack has taken from something installed beside it is named in the pack's own suite.

What it deliberately does not judge is written in its docstring, with a reason for each. The two worth knowing here: a block's summary and a config field's docstring are documentation written beside the value, not refusals, so they are not held to this; and an exception a pack raises for itself and catches before it answers is not a refusal until something catches it and refuses under a code.

A stream a block captures comes through the context

A block that runs something which prints -- a process, a container, a remote job -- never names a storage URI of its own. It asks the context for a capture, writes the stream through it as it arrives, and puts the URI it is handed in its output:

printed = 0
async with ctx.capture("stdout") as sink:
    async for chunk in job.stdout():
        printed += await sink.write(chunk)
return AcmeRunOutput(stdout_uri=sink.uri, stdout_bytes=printed)

The engine names the object under the run's scratch prefix, after the step, the fan-out item and the attempt, so a retry never writes over what the attempt before it printed. The name is what distinguishes one stream from another inside a step: "stdout" and "stderr", or f"{command}-stdout" where a block runs several. content_type defaults to text/plain.

How much of a stream a block may also inline in its output is ctx.inline_capture, in bytes. Inline the head, count the whole, say whether the two disagree, and leave the rest to the URI: a step that prints a gigabyte must be a file in storage, never a gigabyte in the worker.

A sensor keeps its place in a cursor

A sensor's poke runs once and returns: either the observation, which ends the step, or NotYet, which parks the attempt until the next poke is due. A sensor that reads a stream needs somewhere to write down how far it has read, and that is the cursor.

class QueueDepth(Sensor[QueueDepthConfig, QueueDepthOutput]):
    spec = SensorSpec(id="demo.queue_depth", summary="Wait for messages on a queue.", source=True)
    config_model: ClassVar[type[BaseModel]] = QueueDepthConfig
    output_model: ClassVar[type[BaseModel]] = QueueDepthOutput

    async def poke(self, config: QueueDepthConfig, ctx: StepContext) -> QueueDepthOutput | NotYet:
        # None on the first poke, and whatever the last committed NotYet returned after that.
        seen = int((ctx.cursor or {}).get("offset", 0))
        messages = await read_from(config.queue, since=seen)
        if not messages:
            return NotYet(message=f"nothing past offset {seen}")
        if len(messages) < config.min_messages:
            # Keep the ground already covered, so the next poke does not read it again.
            return NotYet(cursor={"offset": seen + len(messages)}, next_poll_in=config.poll_every)
        return QueueDepthOutput(messages=messages, count=len(messages))

Three rules make a cursor safe to rely on. It replaces rather than merges, so everything worth keeping is copied forward into each NotYet, and omitting it leaves the stored one as it is. It is stored by the transaction that parks the attempt, which makes advancing at-least-once: a worker that dies between the read and that commit leaves the older cursor behind, and the next poke reads the same ground again, so a poke must tolerate that. And it lives only as long as the waiting attempt -- a poke that succeeds ends the step, and in a step nothing carries the cursor past it. Anything a downstream step needs belongs in the output.

A sensor may also be poked outside any run, by a sensor source that starts one run per success. It opts in with source=True on its SensorSpec, and the apply refuses a source naming a sensor that has not. Such a sensor reads nothing of a run but ctx.cursor and ctx.started_at, which is when the source began waiting -- the last success, or the first poll -- and a poke that touches scratch, work, runs, capture, a schema or the run id fails as source.sensor_outside_run. It says where a success left off by overriding resume_cursor, which reads it off the output the poke succeeded with. The source stores what it returns and hands it to the first poke of the next wait, so the ground is read on from the end of the last batch rather than succeeding twice; None, the default, starts every wait fresh. In a test, carry_cursor(sensor, answer, ctx) from dirigent-testing hands the next poke what the engine would, after a park or after a success:

def resume_cursor(self, output: QueueDepthOutput) -> JsonMap | None:
    return {"offset": output.messages[-1].offset + 1} if output.messages else None

A formatter is a second extension point

dg format renders a stream, and the formatter it dispatches on is contributed through a second entry point group and a second collecting extension point:

# the pack's pyproject.toml
[project.entry-points."dirigent.formatters"]
shouty = "my_package:plugin"

from typing import Any

from dirigent_common import Formatter
from dirigent_plugin import extension


class Shouty:
    name = "shouty"
    version = "1"

    def render(self, record: dict[str, Any]) -> str:
        return str(record.get("message", "")).upper()


class ShoutyPlugin:
    @extension
    def formatters(self) -> list[Formatter]:
        return [Shouty()]


plugin = ShoutyPlugin()

The CLI owns this manager. It scans the dirigent.formatters group once, calls each formatters(), and indexes every formatter returned by its name. That name is what dg format <name> dispatches on, so two packages claiming one -- or a package claiming the built-in console or compact -- is refused at startup, the way two packs claiming one block id are. The Formatter protocol lives in dirigent_common: a name, a version, and one render method, so a package ships a formatter without depending on the CLI or on the block contract. A formatter renders a record kind it has never heard of rather than failing.

Example shelves are a third extension point

A distribution may carry example documents, and the host reads them through a third collecting extension point on the same dirigent.plugins.v1 group. examples() answers with the directories the shelves live in, as importlib.resources traversables or plain paths:

from collections.abc import Sequence
from importlib.resources import files
from importlib.resources.abc import Traversable

from dirigent_plugin import extension


class ExamplesPlugin:
    @extension
    def examples(self) -> Sequence[Traversable]:
        return [files("my_package") / "shelves"]


plugin = ExamplesPlugin()

A plugin implements whichever of the three points it has something to say through; a distribution that only carries documents implements examples() and no contribute() at all. The host walks each directory recursively on the first call and never at startup, so a worker pays nothing for a corpus it will not read. A file that parses as a dirigent/v1 document becomes one catalogue entry -- its code, name, description, tags, requires, the shelf it sits on, and its text -- and anything else on the shelf is passed over with a warning naming the plugin and the file. Two plugins may carry one code; one plugin carrying it twice keeps the first. The core corpus ships this way, as dirigent-examples.

What the host reads is what every surface serves: GET /api/v1/examples and GET /api/v1/examples/{code}, dg examples list and dg examples show, and dg pipeline new for a document wearing the starter tag. A pack's shelves therefore reach a person the moment the pack is installed, with nothing to register beyond this hook.

The starter rule. A document earns starter by being a real multi-step flow on a real source -- a public endpoint, or a service a connection names -- and it carries its own connections: and schemas: so that dg run --local runs it alone. dg pipeline new <code> copies its text verbatim into pipelines/, rewrites the code: line, drops starter from the tags, and moves each carried section into requires:, so the copy names the connections and schemas to create first instead of carrying them.

A SQL engine is a fourth extension point

The sql family drives any backend a SQLAlchemy async driver reaches. A backend that needs more than a driver -- DuckDB, whose databases are files and which has no async driver at all -- is a package of its own, under a group the family scans for itself:

# the pack's pyproject.toml
[project.entry-points."dirigent.sql.engines.v1"]
duckdb = "dirigent_block_duckdb:plugin"

An engine subclasses SqlEngine from dirigent_block_sql.engines, claims the one backend a URL names it by, and answers five questions: validate refuses a connection it cannot open, resolve says where a database written as a relative path lands, bind says what a parameter becomes before the database sees it, check reaches the database for dg connection check, and session opens what a step runs its statements in. Adding an engine is that contract in full, and dirigent-block-duckdb is the worked example. Nothing in sql.query or sql.execute changes for a new backend.

How a pack connects systems together

The surfaces are the vocabulary; the connecting happens in a pipeline. A connection kind plus the blocks that speak it bring one external system into reach. From there, packs compose without knowing about each other: every block reads ${steps.<name>.output...} and writes a typed output, so an acme export can feed a convert transform, feed an s3 write, feed an http.request post -- three packs meeting only through the run's data plane and the shared step context.

The connection registry is the seam to each system. A step names a connection by code; the connection holds the base URL and the credential. The same document points at staging or production by swapping the connection, not the pipeline. dirigent reaches out to the systems its connections name; nothing registers inbound. "Connecting things" is therefore: install the packs that bring the systems, then wire their blocks in a DAG through connections and ${...} references.

A connection kind may declare the mark it wears

A kind is drawn as well as named: the connections listing, the New connection dialog, the home screen's health rows and the step form's reference row all lead with a glyph. dirigent ships one for each kind it knows, and a kind it does not know is drawn by a neutral dot. A pack that would rather be recognised declares its own:

class AcmeConnectionKind(ConnectionKind):
    id: ClassVar[str] = "acme"
    config_model: ClassVar[type[BaseModel]] = AcmeConnectionConfig
    mark: ClassVar[str | None] = "M12 2 L22 12 L12 22 L2 12 Z"

mark is the d of one SVG path drawn on a 24-unit grid, monochrome: the UI fills it with currentColor at the size every other kind's glyph is drawn at, so it takes the ink of whatever draws it in both palettes. It is checked when the contribution is built -- non-empty, ASCII, path-data characters only, at most 4 kB -- and a pack that declares anything else is refused with plugin.invalid_mark rather than handing a browser markup. assert_contribution_conforms makes the same check, so a pack reads the refusal in its own test run.

A connection kind that takes one of several credentials says which

A config model that takes a token or a basic pair, and refuses to go without one, says so in a model validator, which a client creating a connection cannot read. credential_forms says it where a client can: each form names every config field it needs, and credential_required says whether a working connection needs one of them at all.

from dirigent_plugin import CredentialForm


class AcmeConnectionKind(ConnectionKind):
    id: ClassVar[str] = "acme"
    config_model: ClassVar[type[BaseModel]] = AcmeConnectionConfig
    credential_forms: ClassVar[tuple[CredentialForm, ...]] = (
        CredentialForm(code="api_token", name="Personal access token", fields=("api_token",)),
        CredentialForm(code="basic", name="Username and password", fields=("basic_username", "basic_password")),
    )
    credential_required: ClassVar[bool] = True

The catalog carries both on the kind's entry. dg connection create reads them: at a terminal, with no form given, it asks which one and then every field of it, the plain basic_username as well as the hidden password; in a script, a kind that needs a credential and got none is refused with cli.connection_needs_a_credential, which names each form and the --set flags that give it. The validator still decides what is accepted. assert_contribution_conforms reports a form that names a field the config model does not declare. dirigent's own http kind declares a bearer token, a basic pair and a header credential, none required, and slack an incoming webhook or a bot token and channel, one required.

What ctx.http reads from a connection

ctx.http(ref) builds a client for a connection of any kind, so it reads a config's fields by name rather than by model:

Field	What the client does with it
base_url	The root every request path resolves against
bearer_token	Sent as Authorization: Bearer <token>
api_token	Sent as Authorization: ApiToken <token>, the scheme a personal access token usually needs
api_token_scheme	The scheme api_token is sent under instead of ApiToken
credential	Sent in credential_header, after credential_scheme when one is set, and only to the origin of base_url
credential_header	The header credential is sent in instead of Authorization
credential_scheme	The word sent before credential
credential_query	The query parameter credential is sent as instead of a header, only to the origin of base_url
basic_username, basic_password	The client's HTTP basic auth
verify_tls	Whether certificates are verified
follow_redirects_with_auth	When true, base_url is followed through its redirects once, without a credential, and every request is sent to where it lands
timeout	The timeout applied to every request

bearer_token wins over api_token when a config carries both, credential wins over either in the same header, and basic credentials win over all of them, because httpx applies auth per request and the token is a client header. A connection kind that names none of these fields still gets a client, with no base URL and no credential.

A free-form header map in a block's config or a connection's is stored as written, so it is typed PlainHeaders from dirigent_common. That refuses a name that carries a credential when the config is validated and, for a block, when the document is applied, whether the value is a literal or a reference. Set hide_input_in_errors on the model as well, so a refusal at run time cannot repeat a value a reference resolved to. A credential goes in a CredentialValue field named credential, with credential_header, credential_scheme or credential_query beside it.

A free-form query map is typed PlainQuery, and a URL a person writes PlainUrl, on the same terms: the first refuses a name that carries a credential, the second that and a user or a password as well. A query name is judged by its own rule, because a query is any API's data: page_token, nextToken, sort_key and a bare key pass, as do a presigned URL's X-Amz-* parameters.

A pack in its own repository

dg blocks new <name> writes a pack to start from: dirigent-<name>, with a pyproject.toml declaring the entry point, a plugin object, one operator, and a test that registers the pack through a plugin manager and calls that operator. uv sync && uv run pytest inside it passes from the first minute, so the first edit is a block rather than wiring.

What a pack is called says what it contributes: dirigent-<system> for an adapter pack (dirigent-dhis2), dirigent-block-<family or engine> for blocks (dirigent-block-parquet, dirigent-block-duckdb), dirigent-storage-<backend> (dirigent-storage-s3), and dirigent-notify-<channel> for a channel. A runtime package is dirigent-<role> -- core, server, cli, client, common, plugin, testing -- and dirigent-blocks is the umbrella over the built-in dirigent-block-* families.

Because the seam is one entry point and the contract is one package, a pack is a normal Python distribution that happens to be discovered. Moving one out of this monorepo into its own repo is mechanical:

• What travels with it: the package (its operators, sensors, and connection kind), and its examples -- the documents that exercise its blocks belong with the pack, not with dirigent, because they are meaningless without it installed. A pack's examples are validated in the pack's own CI against the catalog the pack itself contributes.
• What it depends on: dirigent-common and dirigent-plugin, and nothing of dirigent-core; dirigent-testing in its dev group, for the fixtures and the conformance kit. All three come from PyPI pinned to one exact version (dirigent-plugin==<version>), the version of the runtime the pack releases against. That is the entire surface an out-of-repo pack targets, and the reason the move is low-risk: the dependency points one way, at a small, versioned contract.
• How it checks itself: dirigent-testing carries the conformance kit, and neither half of it imports dirigent-core. assert_contribution_conforms(contribution) checks the blocks a contribution provides are well-formed. check_pack_examples(contribution, directory) checks each of the pack's own documents: that it is a dirigent/v1 pipeline coded after its file, that every block it names is one the pack contributes, that each config fits that block's published schema, and that a connection a step names is one the document accounts for -- carried in its own connections:, so the document runs alone under dg run --local, or listed under requires.connections. Both answer with a list of human-readable issues, and an empty list means it passed.
• How it stays compatible: api_version on the Contribution is the runtime handshake, and a pack written against another revision is refused at load rather than half-working. Before 1.0 the contract changes directly: a dirigent release is a pack release at the same version, and dirigent-integration assembles every pack and runs everyone's tests against the new tips before a release is called done.

The result is that a new integration is a repository, an entry point, and an install, developed and tested on its own cadence and wired into any instance the moment it is present.

Using a pack end to end: the Acme adapter

From an empty directory to a running acme pipeline.

1. A project and an instance

uv init flows && cd flows
uv add dirigent-cli dirigent-acme   # the runtime, the built-in blocks and the server, plus the pack
uv run dg init                      # scaffolds an instance, a first admin, and an example document

uv add dirigent-acme is the whole of installing the adapter: the entry point it declares is what the instance discovers at startup, so dg blocks lists the acme.* blocks and acme appears as a connection kind, with nothing registered by hand. A pack added to an instance that is already running is picked up on the next restart.

2. A connection

The credential lives in one connection, created once; documents name it by code and never carry it:

uv run dg connection create acme acme-prod \
  --set base_url=https://api.acme.example/v2 \
  --set basic_username=ops --set basic_password=district
# or a personal access token instead of basic auth:
#   --set api_token=<PAT>
uv run dg connection check acme-prod     # confirms the credential and reports the server version

At a terminal a required field the command was not given is prompted for, and a bare dg connection create acme acme-prod offers every secret the kind declares; in a script every value arrives via --set, and an empty one (--set api_token=) leaves the field unset.

3. Schemas to validate against (optional)

To catch a change in the service's payload at the boundary, hold the expected shape as a schema and gate on it (see JSON Schema):

uv run dg schema create schemas/acme-sites.json

4. A pipeline

A document that reads the site list, checks its shape, and exports a day of orders to storage, naming the connection by code:

format: dirigent/v1
kind: pipeline
code: nightly-export
requires:
  blocks: [acme.sites, validate.schema, acme.orders]
  schemas: [acme-sites]
steps:
  sites:
    block: acme.sites
    config:
      connection: acme-prod
      region: nordics
      fields: id,name,elevation
  check:
    block: validate.schema
    depends_on: [sites]
    config:
      input: ${steps.sites.output.value}
      schema: acme-sites
  export:
    block: acme.orders
    depends_on: [check]
    config:
      connection: acme-prod
      site: NO-BRGN-01
      period: 2026-01
      measure: TEMPERATURE
  store:
    block: storage.write
    depends_on: [export]
    config:
      target: "${run.scratch}/orders.json"
      value: ${steps.export.output.orders}

Apply it, then run it:

uv run dg apply nightly-export.yaml
uv run dg run nightly-export --watch

Standalone, with no server

A document that carries its own connections: block runs on its own, which is how a pack's own examples work. They live with the pack in the pack's own repository; from a checkout of it:

uv run dg run --local examples/acme-orders.yaml

A real pack built exactly this way is dirigent-dhis2. Two more are published the same way, each with its own documentation: dirigent-contrib, the Chap and climate service packs, and dirigent-fhir, a FHIR R4 connection kind with read, search, create, update and transaction blocks, FHIRPath on the transform verbs, CQL and measures, and structural validation. dg init --pack offers all three.