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 retry 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 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."""
        return BucketOutputs(url=f"gs://{self.name}", created_at="...")

    async def on_observe(self) -> BucketOutputs | None:
        """Look the bucket up by identity; None when it does not exist."""
        return BucketOutputs(url=f"gs://{self.name}", created_at="...")

    async def on_update(self, previous_config: BucketConfig | None) -> BucketOutputs:
        """Converge to self.config; previous_config is None when unknown."""
        return BucketOutputs(url=f"gs://{self.name}", created_at="...")

    async def on_delete(self) -> None:
        """Delete the bucket; succeeds when it is already gone."""

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.

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.

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

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