How to Use Prefect Blocks for Storing External Service Credentials Securely

Prefect blocks store API keys and passwords as encrypted values using SecretStr fields, ensuring credentials never appear in plain text in the UI, logs, or CLI while remaining accessible to your flows via the get_client() method or get() accessor.

Prefect blocks in the PrefectHQ/prefect repository provide first-class, versioned configuration objects for managing sensitive infrastructure data. When you adopt Prefect blocks for storing external service credentials, the framework automatically encrypts secret fields at rest and masks them in every interface, eliminating the risk of accidental exposure in version control or execution logs.

How Secret Fields Are Automatically Detected

Prefect's block schema builder automatically identifies sensitive data by walking the Pydantic model's type tree. In src/prefect/blocks/core.py, the _collect_secret_fields method (lines 64-71) scans for pydantic.SecretStr, pydantic.SecretBytes, or the Prefect-provided Secret block type, marking each as secret in the schema.

This automatic detection ensures that any field containing credentials is encrypted by the Prefect server and never rendered in plain text. When the block is displayed in the UI or CLI, these fields appear as "*****" regardless of the actual value (see the masking implementation in src/prefect/blocks/core.py, lines 428-432).

Using the CredentialsBlock Base Class

For external services requiring a client object—such as databases or cloud providers—Prefect provides the abstract CredentialsBlock class defined in src/prefect/blocks/abstract.py (lines 37-64). This base class standardizes credential management by requiring subclasses to implement a get_client method that returns a ready-to-use SDK client.

Subclasses declare secret fields using type hints like username: SecretStr and implement get_client to handle authentication internally. The base class also provides a logger that respects flow run context, ensuring consistent observability across credential operations.

Working with the Built-in Secret Block

For simple key-value storage without custom client logic, the built-in Secret block in src/prefect/blocks/system.py provides a lightweight solution. Lines 72-80 implement the get() method, which returns the decrypted plaintext value when called.

Unlike custom blocks that wrap service clients, the Secret block stores arbitrary strings directly. You persist the value using save() and retrieve it later within tasks or flows, making it ideal for database passwords, API tokens, or encryption keys that don't require additional configuration parameters.

Complete Implementation Guide

Follow this workflow to securely store and retrieve credentials using Prefect blocks.

Define a Custom Credentials Block

Create a subclass of Block that uses SecretStr for sensitive fields and implements service-specific client initialization:

from prefect.blocks.core import Block
from pydantic import Field, SecretStr, HttpUrl

class RestApiCredentials(Block):
    """Block that stores credentials for a generic REST API."""
    base_url: HttpUrl = Field(..., description="Base URL of the API")
    api_key: SecretStr = Field(..., description="API key for authentication")

    def get_client(self):
        """Return a simple HTTP client with the auth header."""
        import httpx
        return httpx.AsyncClient(
            base_url=str(self.base_url),
            headers={"Authorization": f"Bearer {self.api_key.get_secret_value()}"}
        )

Save and Version the Block

Instantiate and persist the block to the Prefect API. The server encrypts the api_key field automatically:

from myblocks import RestApiCredentials

creds = RestApiCredentials(
    base_url="https://api.example.com",
    api_key="super-secret-token"
)
creds.save("my-api-credentials", overwrite=True)

Load and Use in Flows

Retrieve the block within a flow or task context. Call get_client() to access the authenticated service client, or use get_secret_value() to extract the raw credential:

from prefect import flow, task
from myblocks import RestApiCredentials

@task
async def fetch_data():
    creds = RestApiCredentials.load("my-api-credentials")
    client = await creds.get_client()
    resp = await client.get("/data")
    return resp.json()

@flow
def my_flow():
    data = fetch_data()
    print(data)

Store Arbitrary Secrets Directly

For simple secret storage without custom logic, use the built-in Secret block:

from prefect.blocks.system import Secret

# Save a secret value

Secret(value="my-db-password").save("db-password", overwrite=True)

# Retrieve it later

pwd = Secret.load("db-password").get()   # returns the plain string

Summary

• Automatic encryption: Prefect detects SecretStr, SecretBytes, and Secret types via _collect_secret_fields in src/prefect/blocks/core.py, ensuring server-side encryption at rest.
• Secure display: Secret values are masked as "*****" in the UI, CLI, and logs (see lines 428-432 of the core block implementation).
• Standardized interface: The CredentialsBlock abstract class in src/prefect/blocks/abstract.py provides a consistent get_client pattern for service authentication.
• Flexible retrieval: Use Block.load() to fetch credentials in flows, then call get_secret_value() or get() (for the built-in Secret block) to access decrypted plaintext only when needed.

Frequently Asked Questions

How does Prefect encrypt block secrets at rest?

Prefect encrypts secret fields server-side using industry-standard encryption algorithms before persisting them to the database. When you save a block containing SecretStr or SecretBytes fields, the plaintext never leaves your environment unencrypted; the Prefect server handles the encryption immediately upon receipt. The decryption key is managed by the Prefect backend, and plaintext is only transmitted back to the client when explicitly requested via methods like get_secret_value() or Secret.get().

Can I rotate credentials stored in a Prefect block?

Yes, credential rotation is supported through the block versioning system. You can update an existing block by calling save() with overwrite=True and providing the new credential value. Prefect maintains version history for blocks, allowing you to track changes while ensuring that active flow runs continue using the version they initially loaded. For zero-downtime rotation, create a new block with a different name, update your flow code to reference the new block, then delete the old block once all running flows complete.

What is the difference between SecretStr and the Secret block?

SecretStr is a Pydantic type used within custom block classes to mark individual string fields as sensitive, while the Secret block is a complete, built-in block implementation in src/prefect/blocks/system.py designed for storing single arbitrary values. Use SecretStr when building custom credential blocks that include multiple configuration fields (like URLs, usernames, and API keys), and use the Secret block when you simply need to store and retrieve a single password or token without additional metadata.

How do I prevent secrets from appearing in Prefect flow logs?

Prefect automatically masks secret fields in logs, task inputs, and flow run outputs. The block schema marks fields as secret during definition (as implemented in src/prefect/blocks/core.py), causing the logging infrastructure to replace the actual value with "*****" in any serialized representation. To ensure complete protection, always use SecretStr or the Secret block type rather than standard strings, and avoid calling get_secret_value() inside print statements or logging calls that might transmit the plaintext to external systems.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}