Temporal

The Temporal package mints a fresh Keycard token for every activity execution. An activity declares the Resources it needs with @grant(...), the worker’s KeycardInterceptor mints a token for each of them before the activity body runs, and access() returns them inside the activity. Nothing is written to workflow history.

• Temporal activities that call external APIs and need per-execution credentials
• Getting database URIs and API keys out of a worker’s environment and into vaulted Keycard Resources
• Durable agents built on Temporal’s OpenAI Agents plugin, with the model key minted from a vault instead of OPENAI_API_KEY

pip install keycardai-temporal

# With the OpenAI Agents model provider (see below)
pip install "keycardai-temporal[openai-agents]" "temporalio[openai-agents]"

keycardai.temporal

keycardai.temporal.openai_agents (requires the openai-agents extra)

from datetime import timedelta

from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.worker import Worker

from keycardai.temporal import KeycardInterceptor, access, grant

LEDGER = "https://ledger.<your-domain>"


@grant(LEDGER)
@activity.defn
async def post_entry(order_id: str) -> str:
    token = access().access_token  # fresh for this execution only
    ...  # call the ledger with the token
    return "posted"


@workflow.defn
class SettlementWorkflow:
    @workflow.run
    async def run(self, order_id: str) -> str:
        return await workflow.execute_activity(
            post_entry, order_id, start_to_close_timeout=timedelta(seconds=30)
        )


async def main() -> None:
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="settlement",
        workflows=[SettlementWorkflow],
        activities=[post_entry],
        interceptors=[KeycardInterceptor("https://<zone-id>.keycard.cloud")],
    )
    await worker.run()

The interceptor mints before the body runs and there is no fallback path: a mint failure fails the activity through Temporal’s normal error channel, so the retry policy you already wrote governs what happens next. See Failure Semantics.

access() returns whatever the Resource’s credential provider produces:

The first two work under a plain @grant(resource): the Application acts as itself and nothing exchanges. Brokered credentials come from a user’s consent, so those Resources need a user subject: declare one with the on-behalf-of or impersonation forms in Identity Modes, and the token delivered into the activity is the provider’s own. The GitHub and Google delegated access guides cover the Provider setup.

With no credential argument, KeycardInterceptor calls keycardai.oauth.server.discover_credential(), the SDK-wide environment convention:

• KEYCARD_CLIENT_ID and KEYCARD_CLIENT_SECRET together build a ClientSecret.
• A token file named by KEYCARD_EKS_WORKLOAD_IDENTITY_TOKEN_FILE, AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE, AWS_WEB_IDENTITY_TOKEN_FILE, or AZURE_FEDERATED_TOKEN_FILE builds a WorkloadIdentity.
• KEYCARD_APPLICATION_CREDENTIAL_TYPE (client_secret or workload_identity) names the type to use, and wins over everything else in the environment.

When the environment can build more than one credential and the type variable does not choose between them, the worker fails at startup with GrantConfigurationError instead of guessing. EKS IRSA injects AWS_WEB_IDENTITY_TOKEN_FILE into pods automatically, so a worker meant to use a client secret on EKS must set KEYCARD_APPLICATION_CREDENTIAL_TYPE=client_secret. Passing any keycardai.oauth.server.ApplicationCredential explicitly skips discovery entirely.

Discovery reads the process environment, not a .env file. If your app loads settings through a library such as pydantic-settings, either export the variables into the environment or build the credential from your settings object and pass it explicitly:

from keycardai.oauth.server import ClientSecret
from keycardai.temporal import KeycardInterceptor

interceptor = KeycardInterceptor(
    "https://<zone-id>.keycard.cloud",
    credential=ClientSecret((settings.keycard_client_id, settings.keycard_client_secret)),
)

To remove the worker’s own secret as well, register a workload identity Provider and let discovery build a WorkloadIdentity from the platform’s runtime token. Run on Fly.io without secrets walks through the setup. One caveat: app-as-itself grants and the OpenAI Agents provider require a ClientSecret credential today, so workload identity covers the on-behalf-of and impersonation modes.

In your Keycard Zone, once:

1. Create an Application for the worker with a credential (a ClientSecret, or any workload identity credential the worker’s platform supports).

2. Create the Resource the activities will call. For plain client-credentials issuance, the Zone Provider is enough as its credential provider: nothing exchanges from its tokens.

3. Add the Resource to the Application’s dependencies. App-only issuance needs the dependency; with no user there is no consent step.

4. For on-behalf-of activities only: the exchange rule requires the exchanging Application to provide the Resource the subject token is audienced to, so the worker’s Application needs a small anchor Resource of its own, with user sessions audienced to it.

@grant("res-a", "res-b") takes any number of Resources as positional arguments (at least one; duplicates are rejected at decoration time). Every declared Resource is minted before the body runs, under the one identity the grant declares.

GITHUB = "https://api.github.com"
GOOGLE_API = "https://www.googleapis.com"


@grant(GITHUB, GOOGLE_API, subject_from="user_id")
@activity.defn
async def sync_issues(user_id: str, repo: str) -> str:
    github_token = access(GITHUB).access_token
    google_token = access(GOOGLE_API).access_token
    ...  # call both APIs as the user
    return "synced"

Both Resources here are brokered Provider-backed, so the grant declares a user subject and each token is exchanged for that user (subject_from is covered in Identity Modes). Vault-backed and Zone Provider Resources combine in a grant the same way.

Inside the body, access() with no argument returns the token of a single-resource grant; under a multi-resource grant it raises GrantConfigurationError, and access(resource) selects the token by name. Minting is all-or-nothing: the body never runs with partial credentials. If any declared Resource fails permanently, the activity fails with the non-retryable KeycardAccessDenied error naming that Resource, even when the others minted. Otherwise the first transient failure raises the retryable KeycardMintFailed naming its Resource, and the activity retry policy governs what happens next.

The identity mode is per activity: one subject applies to every Resource in the grant.

• @grant(resource, ...): the Application acts as itself (client credentials). Requires a ClientSecret credential.
• @grant(resource, ..., subject_from=...): the Application acts on behalf of a user. The activity input carries an identity reference (a user id, never a token). The interceptor’s subject_token_provider, an application-supplied session lookup, returns that user’s current session token, and an RFC 8693 exchange turns it into a token for each Resource. For a brokered Provider-backed Resource, the exchanged token is the external provider’s own.
• @grant(resource, ..., subject_from=..., impersonate=True): impersonation, for workflows that outlive the user’s session. The located value is a stable user identifier (email or oid) sent directly to the Zone, which mints a short-lived substitute-user token for each Resource. No session lookup runs and no subject_token_provider is needed.

Impersonation is a different trust model from delegation: the worker asserts who the user is, and Zone policy is the control. It requires a confidential client, Application consent set to implicit, each Resource declared as a dependency of the Application, a prior delegated grant established by the user for each Resource, and Zone policy that explicitly permits the Application to impersonate (forbidden by default). Prefer live delegation whenever the user’s session is still expected to exist. See Act on Behalf of Absent Users for the consent setup and policy requirements.

from typing import Annotated
from dataclasses import dataclass

from keycardai.temporal import Subject, grant

GITHUB = "https://api.github.com"


@dataclass
class Order:
    order_id: str
    approver_id: Annotated[str, Subject()]


@grant(GITHUB)                                       # Subject() marker, validated at decoration time
async def approve(order: Order) -> None: ...

@grant(GITHUB, subject_from="approver_id")           # parameter name ...
async def approve(order_id: str, approver_id: str) -> None: ...

@grant(GITHUB, subject_from="order.approver_id")     # ... or a dotted path into one
async def approve(order: dict) -> None: ...

@grant(GITHUB, subject_from=lambda order: order["approver_id"])  # sync callable escape hatch
async def approve(order: dict) -> None: ...

Use one strategy per activity: a Subject() marker together with subject_from is rejected at decoration time.

The worker supplies the session lookup:

async def session_token_for(approver_id: str) -> str:
    return await sessions.current_token(approver_id)

interceptor = KeycardInterceptor(
    "https://<zone-id>.keycard.cloud",
    subject_token_provider=session_token_for,
)

Transient mint failures are retryable; permanent failures are not. Every typed error in keycardai-oauth carries a retryable property, and the interceptor converts it into the matching Temporal behavior:

List "GrantConfigurationError" in the retry policy’s non_retryable_error_types to give up sooner on misdeclarations.

Workflow history is replayable and permanent, so unlike header-based context-propagation interceptors, the interceptor writes nothing to activity headers, arguments, or return values. The token exists only inside one execution’s context. The same rule is why on-behalf-of activities receive an identity reference instead of a token: the session lookup and exchange happen inside the execution, so a session revoked mid-workflow is never replayed from state.

The package’s test suite runs a real workflow against a local Temporal dev server, then scans the recorded history, including base64-decoded payloads, and asserts the minted token appears nowhere.

The interceptor mints per call and caches nothing, per Keycard’s credential rules; it creates one OAuth client per worker and reuses it, so only the tokens are fresh. Async activities work, and so do sync activities on the thread-pool executor, but a process-pool executor does not: contextvars do not cross processes.

temporalio.contrib.openai_agents.OpenAIAgentsPlugin runs every model call as an activity of its own, registered by the plugin before any of your activities exist, so neither @grant nor the interceptor can hand it a credential. keycardai.temporal.openai_agents.KeycardOpenAIProvider fills that seat: a ModelProvider whose OpenAI client resolves its key through Keycard on each request instead of reading OPENAI_API_KEY once at startup.

pip install "keycardai-temporal[openai-agents]" "temporalio[openai-agents]"

The intended end state for a worker: the interceptor once, @grant on each activity, and the plugin configured through Keycard for the model.

from datetime import timedelta

from temporalio.client import Client
from temporalio.contrib.openai_agents import ModelActivityParameters, OpenAIAgentsPlugin
from temporalio.worker import Worker

from keycardai.temporal import KeycardInterceptor
from keycardai.temporal.openai_agents import KeycardOpenAIProvider

ZONE = "https://<zone-id>.keycard.cloud"


async def main() -> None:
    client = await Client.connect(
        "localhost:7233",
        plugins=[
            OpenAIAgentsPlugin(
                model_params=ModelActivityParameters(
                    start_to_close_timeout=timedelta(seconds=60)
                ),
                model_provider=KeycardOpenAIProvider(
                    ZONE, "urn:vault:openai-api-key", refresh=timedelta(minutes=5)
                ),
            )
        ],
    )
    worker = Worker(
        client,
        task_queue="research",
        workflows=[ResearchWorkflow],
        activities=[vector_search, rerank],
        interceptors=[KeycardInterceptor(ZONE)],
    )
    await worker.run()

KeycardOpenAIProvider(zone_url, resource, credential=None, *, refresh=timedelta(minutes=5), base_url=None, use_responses=None) takes the Zone URL and the identifier of the Resource whose vault holds the OpenAI key. The credential is the same as the interceptor’s: an explicit ApplicationCredential, or the one discover_credential() finds in the environment when omitted. It must be a ClientSecret, as for any client-credentials grant; other credential types raise GrantConfigurationError at construction.

The key is minted with a client-credentials grant on the first model call, not at worker startup, and reused for refresh (or the grant’s expires_in, whichever is shorter). The next model call after the window mints again, so a key rotated in Keycard reaches the worker within one refresh window, with no restart. Concurrent model calls on an expired cache share one mint. A permanent grant failure fails the model activity with the same non-retryable KeycardAccessDenied a @grant activity would raise; a transient one is left to the model activity’s retry policy. The plugin requires an explicit start_to_close_timeout or schedule_to_close_timeout whenever a custom provider is set.