Programmatic Deployments with the Pulumi Automation API

This guide shows how to deploy infrastructure entirely from Python using auto.create_or_select_stack, an inline program, and up() — capturing typed outputs and handling failures without ever invoking the CLI. It extends The Pulumi Automation API overview, which is part of the broader Pulumi Patterns & Provider Management workflow. The goal is a single self-contained driver you can drop into a service, a test fixture, or a CI job.

Context

A programmatic deployment matters when infrastructure is triggered by something other than a human at a terminal: a web request that provisions a per-customer environment, an integration test that needs real cloud resources, or a pipeline step that computes configuration at runtime and then applies it. Doing this by shelling out to the pulumi binary forces you to parse stdout and lose typed access to outputs. The Automation API gives you the engine as a library, so a deployment is just a function call that returns structured results.

Prerequisites

  • Python 3.9+ with pulumi>=3.0 installed: pip install pulumi.
  • The Pulumi CLI binary on PATH (the engine the Automation API drives) and a configured backend — verify with pulumi whoami.
  • AWS credentials in the environment (AWS_PROFILE or OIDC) with permission to create the demo resource (an S3 bucket): s3:CreateBucket, s3:DeleteBucket, s3:PutBucketTagging.
  • A passphrase or secrets provider set (e.g. PULUMI_CONFIG_PASSPHRASE) so the stack can encrypt secret config.

Implementation

1. Define the inline program and select the stack

The inline program is a plain function. create_or_select_stack creates the stack if it does not exist or attaches to it if it does, which makes the driver idempotent across runs.

# Run with: python deploy.py
from typing import Callable
import pulumi
import pulumi_aws as aws
from pulumi import automation as auto

def make_program(env: str) -> Callable[[], None]:
    def program() -> None:
        # Provider note: executed by the engine, not at module import time.
        bucket = aws.s3.BucketV2(f"{env}-data", tags={"env": env})
        # State implication: this resource is recorded in the selected stack's checkpoint.
        pulumi.export("bucket_name", bucket.bucket)
    return program

stack = auto.create_or_select_stack(
    stack_name="dev",
    project_name="automation-demo",
    program=make_program("dev"),  # inline program: no project directory
)

2. Configure the workspace and stack

Install the provider plugin into the workspace, then set configuration through typed methods. Secret values are marked so the backend encrypts them.

# python deploy.py
stack.workspace.install_plugin("aws", "v6.0.0")
stack.set_config("aws:region", auto.ConfigValue(value="us-east-1"))
# State implication: secret=True encrypts the value with the stack's secrets provider.
stack.set_config("automation-demo:apiToken", auto.ConfigValue(value="tok_xyz", secret=True))

3. Refresh, deploy, and capture outputs

Optionally refresh to reconcile state with reality, then up. The result object exposes outputs as typed values you return to the caller.

# python deploy.py
from typing import Dict

def deploy(stack: auto.Stack) -> Dict[str, str]:
    stack.refresh(on_output=print)  # State implication: reconciles drift before applying.
    result = stack.up(on_output=print)
    # outputs maps name -> OutputValue; .value is the resolved Python value.
    return {key: out.value for key, out in result.outputs.items()}

outputs = deploy(stack)
print("provisioned bucket:", outputs["bucket_name"])

4. Wrap the run with structured error handling

Engine failures raise typed exceptions. Catch them so a service returns a clean error and a pipeline exits non-zero with a useful message instead of a stack trace.

# python deploy.py
from pulumi.automation.errors import (
    CommandError,
    ConcurrentUpdateError,
    StackAlreadyExistsError,
)

def safe_deploy(stack: auto.Stack) -> int:
    try:
        outputs = deploy(stack)
        print("ok:", outputs)
        return 0
    except ConcurrentUpdateError:
        # State implication: another operation holds the backend lock; do not retry blindly.
        print("stack is locked by another update; aborting")
        return 2
    except CommandError as exc:
        print(f"engine error: {exc}")
        return 1

if __name__ == "__main__":
    raise SystemExit(safe_deploy(stack))

Verification

Assert the operation succeeded and the expected output came back, then confirm the same stack is visible to the CLI on the shared backend.

# python -m pytest test_deploy.py
def test_deploy_returns_bucket_name() -> None:
    result = stack.up(on_output=print)
    assert result.summary.result == "succeeded"
    assert result.outputs["bucket_name"].value.startswith("dev-data")
# The driver and CLI share one backend, so the result is independently checkable.
pulumi stack output bucket_name --stack dev
aws s3 ls | grep dev-data

Gotchas & Edge Cases

Outputs you forgot to export are absent from result.outputs. The Automation API only returns what the program calls pulumi.export on. If result.outputs["bucket_name"] raises KeyError, the export is missing from the inline program, not lost by the API.

Secret outputs are redacted unless you opt in. A secret=True config value flows into a secret output. Reading .value gives you the plaintext in-process, but on_output logs will show [secret]. Never print secret outputs to shared CI logs.

create_or_select_stack is idempotent, but create_stack is not. If you call create_stack on a name that already exists you get StackAlreadyExistsError. Use create_or_select_stack in any driver that may run more than once, such as a retried CI job or a long-lived service.