Pragma SDK

Documentation | CLI | Providers

Build providers and interact with Pragmatiks programmatically.

Quick Start

Synchronous Client

from pragma_sdk import PragmaClient, Resource

with PragmaClient() as client:
    # Apply a resource
    client.apply_resource(
        Resource(
            provider="gcp", resource="storage", name="my-bucket", config={"location": "US", "storage_class": "STANDARD"}
        )
    )

    # Get resource status
    bucket = client.get_resource("gcp", "storage", "my-bucket")
    print(bucket.outputs)

Asynchronous Client

import asyncio
from pragma_sdk import AsyncPragmaClient, Resource


async def main():
    async with AsyncPragmaClient() as client:
        # Apply a resource
        await client.apply_resource(
            Resource(
                provider="gcp",
                resource="storage",
                name="my-bucket",
                config={"location": "US", "storage_class": "STANDARD"},
            )
        )

        # Get resource status
        bucket = await client.get_resource("gcp", "storage", "my-bucket")
        print(bucket.outputs)


asyncio.run(main())

Installation

pip install pragmatiks-sdk

Or with uv:

uv add pragmatiks-sdk

Features

• HTTP Clients - Sync and async clients for the Pragmatiks API
• Provider Authoring - Build custom providers with type-safe Config and Outputs
• Provider Publishing and Installation - Publish provider versions, install, upgrade and downgrade them, and read an installation's history
• Dead-Letter Queue - Inspect and delete failed events for debugging and recovery
• Field References - Reference outputs from other resources dynamically
• Testing Harness - Test provider lifecycle methods locally without deployment
• Auto-discovery - Automatic credential resolution from environment or config files

Building Providers

Define resources with typed configuration and lifecycle methods:

from pragma_sdk import AlreadyExistsError, Provider, Resource, Config, Outputs, Field

gcp = Provider()


class BucketConfig(Config):
    location: Field[str]
    storage_class: Field[str] = "STANDARD"


class BucketOutputs(Outputs):
    url: str
    created_at: str


@gcp.resource("storage")
class Bucket(Resource[BucketConfig, BucketOutputs]):
    async def on_create(self) -> BucketOutputs:
        """Provision the bucket; refuse a bucket that already exists."""
        bucket = await storage_client.create_bucket(self.name, location=self.config.location)

        if bucket is None:
            raise AlreadyExistsError(f"Bucket {self.name} already exists")

        return BucketOutputs(url=f"gs://{self.name}", created_at=bucket.created_at)

    async def on_observe(self) -> BucketOutputs | None:
        """Look the bucket up by identity; None when it does not exist."""
        bucket = await storage_client.find_bucket(self.name)

        if bucket is None:
            return None

        return BucketOutputs(url=f"gs://{self.name}", created_at=bucket.created_at)

    async def on_update(self, previous_config: BucketConfig | None) -> BucketOutputs:
        """Converge to self.config; previous_config is None when unknown."""
        bucket = await storage_client.update_bucket(self.name, storage_class=self.config.storage_class)

        return BucketOutputs(url=f"gs://{self.name}", created_at=bucket.created_at)

    async def on_delete(self) -> None:
        """Delete the bucket; succeeds when it is already gone."""
        await storage_client.delete_bucket(self.name, missing_ok=True)

storage_client stands for your vendor's client.

Observing before acting

Before it creates, updates or deletes a resource of a type that is not computed, the platform calls on_observe to see whether the object exists, and calls it again once the lifecycle method returns:

• A type that is not computed must define on_observe and declare an Outputs class (Resource[YourConfig, YourOutputs]): @provider.resource() refuses it otherwise, with MissingObserveError or MissingOutputsError, since without outputs the platform could never see its object as present. Publishing such a provider fails admission with missing_observe or missing_outputs, naming the resource type.

• on_observe returns the object's outputs, or None only when the object does not exist. Platform callbacks such as apply_resource are not available in it.

• on_create and on_update return only once on_observe finds the object, with outputs equal to what on_observe returns. Both sides are validated against the type's Outputs class before they are compared, so properties it does not declare never count. on_delete returns only once on_observe returns None.

• When the vendor refuses a create because the object already exists, raise AlreadyExistsError from on_create instead of adopting the object. The resource fails, and Pragmatiks never takes over an object it did not create.

• Outputs that fail validation, or that differ from what on_observe returns, fail the resource while the object stays in place, until a fixed provider version runs.

ProviderHarness results carry normalized_outputs, the outputs as the platform compares them, so a test can check that a create and an observe agree.

Declaring the provider

A provider wheel declares itself with one entry point in the pragma.provider group, named after the provider and pointing at its importable package:

[project.entry-points."pragma.provider"]
gcp = "gcp_provider"

The Pragmatiks platform finds the provider's resource types through this entry point. Only resource classes defined inside that package belong to the provider; resource classes it imports from another provider stay that provider's.

A resource always names the provider whose package defines its class. When a provider builds or applies a resource of a type it imports from another provider, that resource, its identity and its owner references name the other provider, and that provider manages it. Declare that provider as a dependency, so it is embedded, and install it in the organization too: the organization's installation of it is what manages those resources.

To check what the platform will read from an installed wheel, run introspection in its environment:

PRAGMA_PROVIDER_DISTRIBUTION=<distribution> python -m pragma_sdk.introspection --output report.json

<distribution> is the name in the wheel's [project] table. When you publish, the wheel is renamed to <organization>-<provider name you publish under> before it is uploaded, and the platform loads it under that name.

The report lists the resource types with their schemas, or why the provider failed to load.

Reading other versions' data

Outputs ignore properties the class does not declare, so outputs a newer version of the provider recorded still read. Config stays strict for the resource being processed. A dependency's config is read with the unknown top-level fields of a strict Config ignored, so a dependency managed by a newer version of its provider still resolves.

Field References

Reference outputs from other resources:

from pragma_sdk import FieldReference

config = AppConfig(
    database_url=FieldReference(provider="postgres", resource="database", name="my-db", field="outputs.connection_url")
)

Testing Providers

Test lifecycle methods locally with ProviderHarness:

from pragma_sdk.provider import ProviderHarness


async def test_bucket_creation():
    harness = ProviderHarness()

    result = await harness.invoke_create(Bucket, name="test-bucket", config=BucketConfig(location="US"))

    assert result.success
    assert "gs://test-bucket" in result.outputs.url

    observed = await harness.invoke_observe(Bucket, name="test-bucket", config=BucketConfig(location="US"))

    assert observed.normalized_outputs == result.normalized_outputs

Inside the harness, a resource of a class from the tested class's top-level package carries the harness's provider_name (test/local by default). When the tested class builds resources of an embedded provider's types, name that provider's package the way the platform does, or reading their provider or id raises:

harness = ProviderHarness(package_catalog_names={"gcp_provider": "pragmatiks/gcp"})

Authentication

Credentials are discovered automatically in this order:

1. Explicit auth_token parameter
2. Context-specific environment variable: PRAGMA_AUTH_TOKEN_{CONTEXT} (e.g., PRAGMA_AUTH_TOKEN_PRODUCTION)
3. Generic environment variable: PRAGMA_AUTH_TOKEN
4. Credentials file: ~/.config/pragma/credentials

The context is determined by: explicit context parameter > PRAGMA_CONTEXT env var > CLI config > "default".

# Auto-discover credentials (uses default context)
client = PragmaClient()

# Explicit token
client = PragmaClient(auth_token="sk_...")

# Use a specific context (checks PRAGMA_AUTH_TOKEN_PRODUCTION first)
client = PragmaClient(context="production")

# Require authentication (fail if no token)
client = PragmaClient(require_auth=True)

API Reference

HTTP Client Methods

Both PragmaClient (sync) and AsyncPragmaClient (async) provide the same methods, except where noted below.

Resources

Resource operations live on a project handle: client.project("my-project").list_resources().

Lifecycle Events

On the client itself, AsyncPragmaClient only.

Providers

Installed Providers

Dead-Letter Queue

User & Health

Provider Classes

Development

# Install the toolchain, dependencies and git hooks
mise run install

# Format code
mise run format

# Type check and lint
mise run check

License

MIT