SDK Overview - Pragmatiks

The Pragmatiks SDK provides Python clients for managing resources programmatically. It offers both synchronous and asynchronous clients, typed responses, and utilities for building providers.

Installation

uv add pragmatiks-sdk

pip install pragmatiks-sdk

Quick Start

The client auto-discovers credentials from pragma auth login or the environment. Resource operations live on a project handle:

from pragma_sdk import PragmaClient

with PragmaClient() as client:
    project = client.project("my-project")
    resources = project.list_resources()
    agents = project.list_resources(provider="pragmatiks/agno", resource="agent")
    agent = project.wait_ready("pragmatiks/agno", "agent", "support-bot")

To create or update resources from YAML, use pragma resources apply.

Authentication

The SDK discovers credentials in this order:

Explicit token - Pass auth_token to the client
Context-specific environment variable - PRAGMA_AUTH_TOKEN_{CONTEXT} (uppercase)
Generic environment variable - PRAGMA_AUTH_TOKEN
CLI credentials file - Tokens stored by pragma auth login

Call	Behavior
`PragmaClient()`	Auto-discovery (recommended)
`PragmaClient(auth_token="your-token-here")`	Explicit token
`PragmaClient(context="production")`	Token for a named context
`PragmaClient(require_auth=True)`	Raises `ValueError` when no token is found
`PragmaClient(auth_token=None)`	No authentication

Contexts

Contexts switch between environments, such as development and production. PragmaClient(context="production") reads PRAGMA_AUTH_TOKEN_PRODUCTION or the credentials file. Set the default context via environment variable:

export PRAGMA_CONTEXT=production

PragmaClient

Synchronous client for the Pragmatiks API. Use as a context manager to ensure connections are closed.

Constructor

class PragmaClient:
    def __init__(
        self,
        base_url: str | None = None,
        timeout: float = 30.0,
        auth_token: str | None = ...,
        context: str | None = None,
        require_auth: bool = False,
    ) -> None: ...

Parameter	Description
`base_url`	API URL; defaults to `PRAGMA_API_URL`
`timeout`	Request timeout in seconds
`auth_token`	Bearer token; omit for auto-discovery
`context`	Named context for token lookup
`require_auth`	Raise when no token is found

Context Manager

The with block closes the connection on exit. Without it, call close():

client = PragmaClient()
try:
    resources = client.project("my-project").list_resources()
finally:
    client.close()

Health Check

def is_healthy() -> bool: ...

Check if the API is reachable and healthy.

with PragmaClient() as client:
    if client.is_healthy():
        print("API is available")
    else:
        print("API is unavailable")

Get Current User

def get_me() -> UserInfo: ...

Get information about the authenticated user.

from pragma_sdk import PragmaClient, UserInfo

with PragmaClient() as client:
    user = client.get_me()
    print(f"User: {user.email}")
    print(f"Organization: {user.organization_name}")

Returns: UserInfo with user_id, email, organization_id, and organization_name. Raises: httpx.HTTPStatusError if not authenticated (401).

list_resource_schemas

def list_resource_schemas(provider: str | None = None) -> list[ResourceSchema]: ...

List the resource types that installed providers register.

for schema in client.list_resource_schemas(provider="pragmatiks/agno"):
    print(f"{schema.provider}/{schema.resource}: {schema.description}")

Returns: List of ResourceSchema with provider, resource, config_schema, outputs_schema, and description.

Resource Operations

client.project(project_id) returns a ProjectResources handle. Every resource operation is scoped to that project.

project = client.project("my-project")

list_resources

def list_resources(
    provider: str | None = None,
    resource: str | None = None,
    tags: list[str] | None = None,
    *,
    model: type[ResourceT] | None = None,
    reveal: bool = False,
) -> list[ResourceT] | list[dict[str, Any]]: ...

List resources, filtered by provider, resource type, and tags (a resource must match every tag). Returns dictionaries by default, or instances of model. reveal=True includes sensitive values.

resources = project.list_resources()
agents = project.list_resources(provider="pragmatiks/agno", resource="agent")
production = project.list_resources(tags=["env:production"])

get_resource

def get_resource(
    provider: str,
    resource: str,
    name: str,
    *,
    model: type[ResourceT] | None = None,
    reveal: bool = False,
) -> ResourceT | dict[str, Any]: ...

Get a single resource by its identifier.

agent = project.get_resource("pragmatiks/agno", "agent", "support-bot")
print(agent["lifecycle_state"])

Raises: httpx.HTTPStatusError if the resource is not found (404).

apply_resource

def apply_resource(
    resource: ResourceT,
    *,
    model: type[ResourceT] | None = None,
    reveal: bool = False,
) -> ResourceT | dict[str, Any]: ...

Create a resource or update an existing one. The resource’s project_id must match the handle’s project. Raises: ProjectMismatchError if resource.project_id differs from the handle’s project; httpx.HTTPStatusError if the API rejects the resource.

deactivate_resource

def deactivate_resource(
    provider: str,
    resource: str,
    name: str,
    *,
    dry_run: bool = False,
) -> TeardownResponse: ...

Tear down the resource’s external object and return the resource to draft. dry_run=True previews the teardown without changing anything.

delete_resource

def delete_resource(
    provider: str,
    resource: str,
    name: str,
    *,
    dry_run: bool = False,
) -> TeardownResponse: ...

Delete a resource. The resource enters the deletion lifecycle and is removed after cleanup.

preview = project.delete_resource("pragmatiks/agno", "agent", "support-bot", dry_run=True)
for impact in preview.impact:
    print(f"{impact.provider}/{impact.resource}/{impact.name}: {impact.action}")

project.delete_resource("pragmatiks/agno", "agent", "support-bot")

Returns: TeardownResponse with lifecycle_state, impact (every resource the teardown reaches, the requested one first), and dry_run.

wait_ready, wait_deactivated, wait_deleted

def wait_ready(
    provider: str,
    resource: str,
    name: str,
    *,
    timeout: float | None = 300.0,
    poll_interval: float = 2.0,
) -> dict[str, Any]: ...

Poll a resource until it reaches ready. wait_deactivated waits for draft and wait_deleted for deleted, with the same parameters. timeout=0 or None waits forever. Raises: ResourceFailedError if the resource reaches failed; TimeoutError if the timeout expires first.

Dead Letter Operations

Dead letter events are failed lifecycle events that can be retried or discarded.

list_dead_letter_events

def list_dead_letter_events(provider: str | None = None) -> list[dict[str, Any]]: ...

List failed events awaiting manual intervention, optionally for one provider.

for event in client.list_dead_letter_events(provider="pragmatiks/agno"):
    print(f"{event['resource_id']}: {event['error']}")

get_dead_letter_event

def get_dead_letter_event(event_id: str) -> dict[str, Any]: ...

Get details of a specific dead letter event.

event = client.get_dead_letter_event("evt_abc123")
print(event["error"])
print(event["payload"])

retry_dead_letter_event

def retry_dead_letter_event(event_id: str) -> None: ...

Retry a single failed event.

client.retry_dead_letter_event("evt_abc123")

retry_all_dead_letter_events

def retry_all_dead_letter_events() -> int: ...

Retry all dead letter events. Returns the count of events retried.

count = client.retry_all_dead_letter_events()
print(f"Retried {count} events")

delete_dead_letter_event

def delete_dead_letter_event(event_id: str) -> None: ...

Discard a dead letter event without retrying.

client.delete_dead_letter_event("evt_abc123")

delete_dead_letter_events

def delete_dead_letter_events(provider: str | None = None, *, all: bool = False) -> int: ...

Bulk delete dead letter events for one provider, or every event with all=True. Returns the count deleted.

count = client.delete_dead_letter_events(provider="pragmatiks/agno")
count = client.delete_dead_letter_events(all=True)

Raises: ValueError if neither provider nor all=True is specified.

Provider Operations

These methods back the pragma providers commands; see the CLI reference for the workflow. Provider names are namespaced as org/name.

Method	Description
`list_providers(query=None, scope=None, tags=None, limit=20, offset=0)`	Browse the provider catalog; returns `PaginatedResponse[Provider]`
`get_provider(name)`	Get a catalog entry
`list_provider_versions(name)`	List the published versions of a provider
`publish_provider_version(*, wheel_path, name, version, schemas, metadata, changelog=None)`	Upload a wheel as a new version; raises `ProviderVersionConflictError` when the version already exists
`install_provider(provider_name, version=None, upgrade_policy=UpgradePolicy.MANUAL, config=None)`	Install a provider in your organization
`list_installations()`	List installed providers
`upgrade_provider(name, target_version=None)`	Upgrade an installed provider
`downgrade_provider(name, target_version)`	Move an installed provider to an earlier version
`uninstall_provider(name, *, cascade=False)`	Uninstall a provider; `cascade=True` deletes its resources
`deploy_provider(provider_name, version=None)`	Redeploy an installed provider; returns `DeploymentResult`
`get_deployment_status(provider_name)`	Get the deployment status; returns `DeploymentResult`

from pragma_sdk import DeploymentStatus

with PragmaClient() as client:
    client.install_provider("pragmatiks/agno")
    result = client.get_deployment_status("pragmatiks/agno")
    if result.status == DeploymentStatus.AVAILABLE:
        print(f"Version {result.version} is running")

DeploymentResult.status is one of pending, progressing, available, or failed.

AsyncPragmaClient

Asynchronous client with the same API as PragmaClient. All methods are async, and client.project(project_id) returns an AsyncProjectResources handle.

import asyncio

from pragma_sdk import AsyncPragmaClient


async def main() -> None:
    """List agents and wait for one to become ready."""
    async with AsyncPragmaClient() as client:
        project = client.project("my-project")
        agents = await project.list_resources(provider="pragmatiks/agno", resource="agent")
        agent = await project.wait_ready("pragmatiks/agno", "agent", "support-bot")


asyncio.run(main())

Async Context Manager

The async with block closes the connection on exit. Without it, await close():

client = AsyncPragmaClient()
try:
    resources = await client.project("my-project").list_resources()
finally:
    await client.close()

Concurrent Operations

Run independent operations concurrently:

import asyncio


async def list_all_agents(project_ids: list[str]) -> list[dict]:
    """List the agents of several projects concurrently."""
    async with AsyncPragmaClient() as client:
        results = await asyncio.gather(
            *(
                client.project(project_id).list_resources(provider="pragmatiks/agno", resource="agent")
                for project_id in project_ids
            )
        )
    return [agent for agents in results for agent in agents]

Lifecycle Events

AsyncPragmaClient.stream_lifecycle_events(timeout=None) yields one frame per resource lifecycle transition in your organization. timeout=0 or None streams until you stop consuming.

Typed Responses

Pass a provider’s resource class as model to get typed instances back. The response’s config and outputs are validated against that class, so use the class from the installed provider version:

from pragma_sdk import LifecycleState

from mycloud_provider import Database

project = client.project("my-project")
database = project.get_resource("myorg/mycloud", "database", "analytics", model=Database)
if database.lifecycle_state == LifecycleState.READY:
    print(database.outputs.connection_url)

databases = project.list_resources(provider="myorg/mycloud", resource="database", model=Database)

Lifecycle States

LifecycleState is a string enum, so it compares equal to the lifecycle_state string in a dictionary response:

State	Meaning
`draft`	Declared but not submitted, or deactivated
`waiting`	Waiting for a dependency to become ready
`pending`	Queued for processing
`processing`	Being created, updated, or deleted
`ready`	Provisioned
`failed`	Failed; `error` holds the reason
`deleting`	Being torn down
`deleted`	Deleted

Error Handling

The SDK raises httpx.HTTPStatusError for API errors. Handle common cases:

import httpx

from pragma_sdk import PragmaClient

with PragmaClient() as client:
    try:
        resource = client.project("my-project").get_resource("pragmatiks/agno", "agent", "missing")
    except httpx.HTTPStatusError as error:
        if error.response.status_code == 404:
            print("Resource not found")
        elif error.response.status_code == 401:
            print("Authentication failed")
        elif error.response.status_code == 403:
            print("Permission denied")
        else:
            raise

Authentication Errors

PragmaClient(require_auth=True) raises ValueError when no credentials are found. Run pragma auth login or set PRAGMA_AUTH_TOKEN.

Timeout Handling

Requests that exceed the client’s timeout raise httpx.TimeoutException. Raise the limit for slow calls:

with PragmaClient(timeout=60.0) as client:
    resources = client.project("my-project").list_resources()

File Uploads

Upload files to Pragmatiks storage programmatically:

with open("training-data.jsonl", "rb") as file:
    result = client.upload_file(
        name="training-data",
        content=file.read(),
        content_type="application/jsonl",
    )

print(result["url"])
print(result["public_url"])
print(result["size"])
print(result["checksum"])

Returns: A dictionary with url, public_url, size, content_type, checksum, and uploaded_at.

Dependencies and Field References

A resource config points at another resource in one of two ways:

Field reference: a single output value, written in YAML as provider, resource, name, and field. See Resources.
Dependency[T]: a whole resource, with typed access to its config and outputs in provider code.

Owner references, which make a child resource deleted along with its owner, are described in Resources.

Dependency[T]

Provider authors declare a whole-resource dependency with Dependency[T]. The platform waits for the dependency to be ready before it runs the dependent’s handlers, and resolve() returns it:

from pragma_sdk import Config, Dependency, Field, Outputs, Resource

from mycloud_provider.resources import Database


class AppConfig(Config):
    """Configuration of an application.

    Attributes:
        database: Database the application connects to.
        image: Container image to run.
    """

    database: Dependency[Database]
    image: Field[str]


class AppOutputs(Outputs):
    """Outputs of an application.

    Attributes:
        database_url: Connection string the application uses.
    """

    database_url: str


class App(Resource[AppConfig, AppOutputs]):
    """Application connected to a managed database.

    Only ``on_create`` is shown; the other handlers follow the same pattern.
    """

    async def on_create(self) -> AppOutputs:
        """Deploy the application against the resolved database.

        Returns:
            Outputs of the application.
        """
        database = await self.config.database.resolve()
        return AppOutputs(database_url=database.outputs.connection_url)

Provider Authoring

The SDK also provides utilities for building providers. See Building Providers for complete documentation.

Key Exports

pragma_sdk exports the building blocks of a resource: the Resource, Config, and Outputs base classes, the Provider registry, the field types, and HealthStatus.

from pragma_sdk import (
    Config,
    Dependency,
    Field,
    FieldReference,
    HealthStatus,
    ImmutableField,
    LifecycleState,
    Outputs,
    Provider,
    Resource,
    SensitiveField,
)

Quick Example

import asyncio

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

from mycloud_client import CloudClient, NotFound

mycloud = Provider()


class DatabaseConfig(Config):
    """Configuration for a managed database.

    Attributes:
        name: Database name. Locates the database, so it cannot change.
        size_gb: Storage size in gigabytes.
    """

    name: ImmutableField[str]
    size_gb: Field[int] = 10


class DatabaseOutputs(Outputs):
    """Outputs of a managed database.

    Attributes:
        connection_url: Connection string for the database.
    """

    connection_url: str


@mycloud.resource("database")
class Database(Resource[DatabaseConfig, DatabaseOutputs]):
    """Managed database located by its name."""

    async def on_observe(self) -> DatabaseOutputs | None:
        """Report the database named by config.

        Returns:
            Outputs for the database, or ``None`` when it does not exist.
        """
        try:
            database = await asyncio.to_thread(CloudClient().get_database, self.config.name)
        except NotFound:
            return None
        return DatabaseOutputs(connection_url=database.connection_url)

    async def on_create(self) -> DatabaseOutputs:
        """Create the database.

        Returns:
            Outputs for the new database.
        """
        database = await asyncio.to_thread(
            CloudClient().create_database, self.config.name, self.config.size_gb
        )
        return DatabaseOutputs(connection_url=database.connection_url)

    async def on_update(self, previous_config: DatabaseConfig | None) -> DatabaseOutputs:
        """Resize the database to the configured size.

        Args:
            previous_config: Configuration last applied, or ``None`` when the
                platform holds none for an existing database.

        Returns:
            Outputs for the resized database.
        """
        database = await asyncio.to_thread(
            CloudClient().resize_database, self.config.name, self.config.size_gb
        )
        return DatabaseOutputs(connection_url=database.connection_url)

    async def on_delete(self) -> None:
        """Delete the database. Succeeds when it is already gone."""
        try:
            await asyncio.to_thread(CloudClient().delete_database, self.config.name)
        except NotFound:
            pass

Next Steps

Building Providers

Create custom providers with lifecycle methods.

Lifecycle Methods

Deep dive into on_observe, on_create, on_update, and on_delete.

API Reference

Full REST API documentation.

Error Recovery

Handle failures and retry dead letter events.

​Installation

​Quick Start

​Authentication

​Contexts

​PragmaClient

​Constructor

​Context Manager

​Health Check

​Get Current User

​list_resource_schemas

​Resource Operations

​list_resources

​get_resource

​apply_resource

​deactivate_resource

​delete_resource

​wait_ready, wait_deactivated, wait_deleted

​Dead Letter Operations

​list_dead_letter_events

​get_dead_letter_event

​retry_dead_letter_event

​retry_all_dead_letter_events

​delete_dead_letter_event

​delete_dead_letter_events

​Provider Operations

​AsyncPragmaClient

​Async Context Manager

​Concurrent Operations

​Lifecycle Events

​Typed Responses

​Lifecycle States

​Error Handling

​Authentication Errors

​Timeout Handling

​File Uploads

​Dependencies and Field References

​Dependency[T]

​Provider Authoring

​Key Exports

​Quick Example

​Next Steps