Best Practices

This guide covers best practices for building production-quality providers. Follow these guidelines to create providers that are reliable, secure, and easy to maintain.

Idempotency

Idempotency is the most critical requirement for providers. Lifecycle methods may be called multiple times with the same input due to retries, restarts, or redelivery.

See Lifecycle Methods - Idempotency Requirements for patterns on implementing idempotent observe, create, update, and delete operations. Key principles:

Observe: Locate the object from identity and config alone; return None when it does not exist
Create: Handle “already exists” errors by retrieving the existing resource
Update: Converge to the full desired state; when previous_config is None, apply everything
Delete: Treat “not found” as success, not failure

Error Handling

Use Specific Exception Types

Choose exception types that communicate the nature of the failure: RuntimeError for infrastructure and API failures, ValueError for user configuration errors:

async def on_create(self) -> DatabaseOutputs:
    try:
        database = await self.client.create_database(self.config.name)
    except QuotaExceededError as error:
        raise RuntimeError(f"Quota exceeded: {error}") from error
    except InvalidNameError as error:
        raise ValueError(f"Invalid database name: {error}") from error

    return DatabaseOutputs(connection_url=database.url)

Exception Type	When to Use	Retryable
`ValueError`	Invalid configuration (user error)	No
`RuntimeError`	Infrastructure or API failures	Depends
Unhandled exceptions	Unexpected errors (preserve traceback)	Yes

Let Unexpected Errors Propagate

Do not catch all exceptions. Handle only the errors you expect and let everything else propagate with its full traceback for debugging:

async def on_create(self) -> SecretOutputs:
    try:
        secret = await self.client.create_secret(self.config.secret_id)
    except AlreadyExists:
        secret = await self.client.get_secret(self.config.secret_id)

    return SecretOutputs(resource_name=secret.name)

Never wrap the call in except Exception and re-raise a generic RuntimeError: it hides the original failure behind a message that says nothing.

Provide Actionable Error Messages

Error messages should help users understand what went wrong and how to fix it. Say which field is wrong and what to change:

raise ValueError("size_gb must be at least 10 for the postgres engine; raise size_gb and apply again")

A generic message such as "Invalid configuration" leaves the user guessing.

Credentials Management

Accept Credentials as Configuration

Providers should accept credentials as configuration, not rely on ambient credentials:

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

    Attributes:
        name: Database name. Locates the database, so it cannot change.
        credentials: Service credentials as a dict or a JSON string, usually a
            reference to a ``pragma/secret`` resource. Never read from the
            environment.
    """

    name: ImmutableField[str]
    credentials: SensitiveField[dict[str, Any] | str]


class Database(Resource[DatabaseConfig, DatabaseOutputs]):
    """Managed database authenticated with the user's credentials."""

    def build_client(self) -> CloudClient:
        """Build a client from the configured credentials.

        Accepts the credentials as a dict or as a JSON string.

        Returns:
            Cloud API client.
        """
        credentials = self.config.credentials
        if isinstance(credentials, str):
            credentials = json.loads(credentials)

        return CloudClient(credentials=credentials)

This approach:

Works in multi-tenant environments where each user has different credentials
Makes credential requirements explicit in the schema
Enables users to manage credentials via Pragma secrets

Use FieldReference for Secrets

Document that users should use FieldReference to pass credentials from a secret resource:

provider: mycloud
resource: database
name: analytics
config:
  name: analytics
  credentials:
    provider: pragma
    resource: secret
    name: mycloud-creds
    field: outputs.credentials

Never Log Credentials

Ensure credentials never appear in logs or error messages. The authentication error from a client library can echo the credentials it was given, so raise a message of your own and drop the original with from None:

def build_client(self) -> CloudClient:
    try:
        return CloudClient(credentials=self.config.credentials)
    except AuthenticationError:
        raise RuntimeError("Authentication failed; check the credentials secret") from None

Never interpolate the caught error into the message.

Performance

Cache Clients Within Lifecycle Methods

Create API clients once per lifecycle invocation, not per operation:

class Database(Resource[DatabaseConfig, DatabaseOutputs]):
    """Managed database."""

    def build_client(self) -> CloudClient:
        """Build a client for one lifecycle call.

        Returns:
            Cloud API client.
        """
        return CloudClient(credentials=self.config.credentials)

    async def on_create(self) -> DatabaseOutputs:
        """Create and configure the database with a single client.

        Returns:
            Outputs of the database.
        """
        client = self.build_client()

        database = await client.create_database(self.config.name)
        await client.configure_database(database.id, self.config.settings)
        await client.enable_monitoring(database.id)

        return DatabaseOutputs(database_id=database.id)

Minimize API Calls in Updates

Only call external APIs for fields that changed. previous_config is None when the object exists but the platform holds no prior configuration; apply the full desired state then:

async def on_update(self, previous_config: DatabaseConfig | None) -> DatabaseOutputs:
    client = self.build_client()
    database = await client.get_database(self.config.name)

    if previous_config is None or previous_config.size_gb != self.config.size_gb:
        await client.resize(database.id, self.config.size_gb)

    if previous_config is None or previous_config.settings != self.config.settings:
        await client.configure(database.id, self.config.settings)

    return DatabaseOutputs(database_id=database.id)

Use Async Where Possible

Lifecycle methods are async. Use async API clients, or run a blocking call with await asyncio.to_thread(...), and run independent calls concurrently with asyncio.gather. A handler that never awaits cannot be cancelled:

async def on_create(self) -> DatabaseOutputs:
    async with AsyncCloudClient(credentials=self.config.credentials) as client:
        database = await client.create_database(self.config.name)
        await asyncio.gather(
            client.configure_settings(database.id, self.config.settings),
            client.setup_monitoring(database.id),
        )

    return DatabaseOutputs(database_id=database.id)

Common Mistakes

Forgetting Idempotency

The most common mistake is assuming lifecycle methods only run once. An on_create that calls create_secret unguarded fails on a retry, because the first attempt already created the secret. Handle “already exists” by reading the secret instead:

async def on_create(self) -> SecretOutputs:
    try:
        secret = await self.client.create_secret(self.config.secret_id)
    except AlreadyExists:
        secret = await self.client.get_secret(self.config.secret_id)
    return SecretOutputs(resource_name=secret.name)

Mutable State Between Calls

Do not store state on the resource instance between lifecycle calls. Each call runs on a fresh instance, so a client that on_create stored on self does not exist when on_update runs. Build what a handler needs inside that handler:

class Database(Resource[DatabaseConfig, DatabaseOutputs]):
    """Managed database."""

    def build_client(self) -> CloudClient:
        """Build a client for one lifecycle call.

        Returns:
            Cloud API client.
        """
        return CloudClient(credentials=self.config.credentials)

    async def on_update(self, previous_config: DatabaseConfig | None) -> DatabaseOutputs:
        """Converge the database with a client built for this call.

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

        Returns:
            Outputs of the converged database.
        """
        client = self.build_client()
        database = await client.update_database(self.config.name, size_gb=self.config.size_gb)
        return DatabaseOutputs(database_id=database.id)

Checking Immutable Fields in Handlers

Fields that cannot change in place, and every field on_observe reads to locate the object, are declared ImmutableField (or ImmutableDependency). The platform refuses a change to them, so a user changes one by deleting and recreating the resource. Never compare these fields against previous_config in a handler:

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

    Attributes:
        name: Database name. Locates the database, so it cannot change.
        region: Region hosting the database. Cannot change in place.
        engine: Database engine. Cannot change in place.
        size_gb: Storage size in gigabytes.
    """

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

Reading self.outputs

self.outputs is never populated when a handler runs. Every handler locates the object from identity (self.name, self.project_id) and self.config, and reads its state from the external system:

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

    Returns:
        Outputs of the database, or ``None`` when it does not exist.
    """
    try:
        database = await self.build_client().get_database(self.config.name)
    except NotFound:
        return None
    return DatabaseOutputs(database_id=database.id, connection_url=database.url)


async def health(self) -> HealthStatus:
    """Report whether the database accepts connections.

    Returns:
        Healthy when the database is ready, degraded otherwise.
    """
    database = await self.build_client().get_database(self.config.name)
    status = "healthy" if database.ready else "degraded"
    return HealthStatus(status=status, details={"ready_replicas": database.ready_replicas})

Outputs hold only stable state: fields derived from identity or config, or read from the object once the write path has waited for it to be ready. Values that change without a platform write, such as replica readiness, counters, or indexing status, belong in health(), never in outputs.

Incomplete Cleanup in on_delete

Delete all resources created by on_create, not just the primary resource. Each step tolerates its object being gone, and the database goes last, because it is what locates the others: a delete interrupted halfway still finds the database on the next attempt and finishes the job:

async def on_delete(self) -> None:
    """Delete the database with its monitoring and backups.

    Every step succeeds when its object is already gone, and the database is
    deleted last, so repeating the delete after a crash finishes the job.
    """
    client = self.build_client()

    try:
        database = await client.get_database(self.config.name)
    except NotFound:
        return

    for delete in (client.delete_monitoring, client.delete_backups, client.delete_database):
        with contextlib.suppress(NotFound):
            await delete(database.id)

Getting Started

Concepts

Resource Catalog

Agno (AI Agents)

GCP

Kubernetes

Building Providers

Idempotency

Error Handling

Use Specific Exception Types

Let Unexpected Errors Propagate

Provide Actionable Error Messages

Credentials Management

Accept Credentials as Configuration

Use FieldReference for Secrets

Never Log Credentials

Performance

Cache Clients Within Lifecycle Methods

Minimize API Calls in Updates

Use Async Where Possible

Common Mistakes

Forgetting Idempotency

Mutable State Between Calls

Checking Immutable Fields in Handlers

Reading self.outputs

Incomplete Cleanup in on_delete

What’s Next

Lifecycle Methods

SDK Reference

​Idempotency

​Error Handling

​Use Specific Exception Types

​Let Unexpected Errors Propagate

​Provide Actionable Error Messages

​Credentials Management

​Accept Credentials as Configuration

​Use FieldReference for Secrets

​Never Log Credentials

​Performance

​Cache Clients Within Lifecycle Methods

​Minimize API Calls in Updates

​Use Async Where Possible

​Common Mistakes

​Forgetting Idempotency

​Mutable State Between Calls

​Checking Immutable Fields in Handlers

​Reading self.outputs

​Incomplete Cleanup in on_delete

​What’s Next

Lifecycle Methods

SDK Reference

Idempotency

Error Handling

Use Specific Exception Types

Let Unexpected Errors Propagate

Provide Actionable Error Messages

Credentials Management

Accept Credentials as Configuration

Use FieldReference for Secrets

Never Log Credentials

Performance

Cache Clients Within Lifecycle Methods

Minimize API Calls in Updates

Use Async Where Possible

Common Mistakes

Forgetting Idempotency

Mutable State Between Calls

Checking Immutable Fields in Handlers

Reading self.outputs

Incomplete Cleanup in on_delete

What’s Next