Create a Custom Provider

In this quickstart, you’ll build a provider that manages webhook endpoints. You’ll define a resource schema, implement lifecycle methods, and publish to Pragmatiks.

Time estimate: 60 minutes. Assumes familiarity with Python and the provider model.

What you’ll build

A webhooks provider with a single resource type: endpoint. Users will be able to create webhook endpoints like this:

provider: webhooks
resource: endpoint
name: deploy-notifications
config:
  url: "https://api.example.com/webhooks"
  events:
    - deployment.created
    - deployment.failed
  secret: "whsec_abc123"

Step 1: Scaffold the project

pragma providers init webhooks
cd webhooks-provider

This creates the project structure:

webhooks-provider/
├── pyproject.toml
├── README.md
└── src/webhooks_provider/
    ├── __init__.py
    └── resources/
        ├── __init__.py
        └── example.py

Step 2: Define the schema

Delete src/webhooks_provider/resources/example.py and create src/webhooks_provider/resources/endpoint.py with your resource definition:

hooks.example.com stands in for the API of the webhooks service you integrate with; it is not a real service. Substitute your service’s endpoint and request format. The handlers show the shape of an observe-then-act provider and work only against a real service.

endpoint.py

import hashlib
from typing import Any

import httpx
from pragma_sdk import Config, Field, Outputs, Resource, SensitiveField

WEBHOOKS_API_URL = "https://hooks.example.com/v1/endpoints"


class EndpointConfig(Config):
    """Configuration for a webhook endpoint.

    Attributes:
        url: URL that receives the webhook deliveries.
        events: Event types delivered to the endpoint.
        secret: Signing secret for deliveries, usually a field reference.
        active: Whether the endpoint receives deliveries.
    """

    url: Field[str]
    events: Field[list[str]]
    secret: SensitiveField[str] | None = None
    active: Field[bool] = True


class EndpointOutputs(Outputs):
    """Outputs exposed to other resources.

    Attributes:
        endpoint_id: Identifier of the endpoint at the webhooks service.
        delivery_url: URL deliveries are sent to.
        active: Whether the endpoint receives deliveries.
    """

    endpoint_id: str
    delivery_url: str
    active: bool


class Endpoint(Resource[EndpointConfig, EndpointOutputs]):
    """Webhook endpoint registered with the webhooks service."""

    def build_endpoint_id(self) -> str:
        """Derive the endpoint identifier from the resource identity.

        The identifier is stable across runs, so ``on_observe`` locates the
        endpoint without stored state.

        Returns:
            Twelve-character identifier unique to this project and name.
        """
        return hashlib.sha256(f"{self.project_id}:{self.name}".encode()).hexdigest()[:12]

    def build_endpoint_url(self) -> str:
        """Build the service URL of this endpoint.

        Returns:
            URL of the endpoint at the webhooks service.
        """
        return f"{WEBHOOKS_API_URL}/{self.build_endpoint_id()}"

    def build_outputs(self, registration: dict[str, Any]) -> EndpointOutputs:
        """Build outputs from the registration the service reports.

        Args:
            registration: Endpoint registration as the service returns it.

        Returns:
            Outputs for the endpoint.
        """
        endpoint_id = self.build_endpoint_id()
        return EndpointOutputs(
            endpoint_id=endpoint_id,
            delivery_url=registration["url"],
            active=registration["active"],
        )

    async def write_registration(self) -> EndpointOutputs:
        """Create or replace the registration with the current configuration.

        Returns:
            Outputs for the registered endpoint.
        """
        async with httpx.AsyncClient() as client:
            response = await client.put(
                self.build_endpoint_url(),
                json={
                    "url": self.config.url,
                    "events": self.config.events,
                    "secret": self.config.secret,
                    "active": self.config.active,
                },
            )
        response.raise_for_status()
        return self.build_outputs(response.json())

    async def on_observe(self) -> EndpointOutputs | None:
        """Report the endpoint registered for this resource.

        Returns:
            Outputs for the endpoint, or ``None`` when it is not registered.
        """
        async with httpx.AsyncClient() as client:
            response = await client.get(self.build_endpoint_url())
        if response.status_code == 404:
            return None
        response.raise_for_status()
        return self.build_outputs(response.json())

    async def on_create(self) -> EndpointOutputs:
        """Register the endpoint.

        Returns:
            Outputs for the registered endpoint.
        """
        return await self.write_registration()

    async def on_update(self, previous_config: EndpointConfig | None) -> EndpointOutputs:
        """Replace the registration with the current configuration.

        The replacement carries the full desired state, so it converges the
        endpoint whether or not ``previous_config`` is known.

        Args:
            previous_config: Configuration last applied, or ``None``.

        Returns:
            Outputs for the registered endpoint.
        """
        return await self.write_registration()

    async def on_delete(self) -> None:
        """Deregister the endpoint. Succeeds when it is already gone."""
        async with httpx.AsyncClient() as client:
            response = await client.delete(self.build_endpoint_url())
        if response.status_code != 404:
            response.raise_for_status()

Key points:

Config defines what users provide. Field[str] means a field that accepts either a direct value or a field reference from another resource.
Outputs defines what this resource exposes. Other resources can reference endpoint_id or delivery_url via field references.
Lifecycle methods are observe-then-act. on_observe reports whether the endpoint exists, locating it from the resource identity and config alone; the platform then calls on_create for an absent endpoint and on_update for a present one. on_observe is mandatory: a resource without it fails to register.
Handlers await their I/O, here through httpx.AsyncClient, and must be idempotent — repeating a handler with the same input produces the same result.

Step 3: Register the resource

Update src/webhooks_provider/resources/__init__.py to export your resource:

__init__.py

from webhooks_provider.resources.endpoint import Endpoint, EndpointConfig, EndpointOutputs

__all__ = ["Endpoint", "EndpointConfig", "EndpointOutputs"]

Update src/webhooks_provider/__init__.py to register it with the provider:

__init__.py

from pragma_sdk import Provider

from webhooks_provider.resources import Endpoint

webhooks = Provider()

webhooks.resource("endpoint")(Endpoint)

Step 4: Publish and install

Publish the provider version, then install it in your organization:

pragma providers publish
pragma providers install <org>/webhooks

publish builds the wheel, extracts the resource schemas, and uploads both under <org>/webhooks, where <org> is your organization’s slug. install queues the provider for your organization, and the platform deploys it.

Step 5: Use it

Now users can create webhook endpoints as Pragmatiks resources:

webhook.yaml

provider: webhooks
resource: endpoint
name: deploy-notifications
config:
  url: "https://api.example.com/webhooks"
  events:
    - deployment.created
    - deployment.failed
  secret:
    provider: pragma
    resource: secret
    name: webhook-secret
    field: outputs.WEBHOOK_SECRET

The secret field uses a field reference — it pulls the value from a pragma/secret resource instead of hardcoding it. Other resources can reference this webhook’s outputs:

config:
  notification_url:
    provider: webhooks
    resource: endpoint
    name: deploy-notifications
    field: outputs.delivery_url

Deep dive

The Building Providers guide covers everything in detail:

Getting Started

Full walkthrough of building a provider from scratch.

Config and Outputs

Schema design with validation, dependencies, and field references.

Lifecycle Methods

Implementing idempotent observe, create, update, and delete handlers.

Deployment

Publishing, installing, and deploying your provider.

Best Practices

Idempotency patterns, error handling, and production readiness.

Getting started

Concepts

Guides

Troubleshooting

Create a Custom Provider

What you’ll build

Step 1: Scaffold the project

Step 2: Define the schema

Step 3: Register the resource

Step 4: Publish and install

Step 5: Use it

Deep dive

Getting Started

Config and Outputs

Lifecycle Methods

Deployment

Best Practices

​What you’ll build

​Step 1: Scaffold the project

​Step 2: Define the schema

​Step 3: Register the resource

​Step 4: Publish and install

​Step 5: Use it

​Deep dive

Getting Started

Config and Outputs

Lifecycle Methods

Deployment

Best Practices

What you’ll build

Step 1: Scaffold the project

Step 2: Define the schema

Step 3: Register the resource

Step 4: Publish and install

Step 5: Use it

Deep dive