Terraform Sdk

Terraform/OpenTofu SDK & API Integration

Integrates with Terraform and OpenTofu using the HCP Terraform API (via pyTFE), the CDKTF Python SDK, and the Terraform Cloud/Enterprise REST API to programmatically manage providers, resources, state, workspaces, and modules.

TL;DR for Code Generation

• Use pyTFE (pytfe.TFEClient) for HCP Terraform and Terraform Enterprise API operations
• For infrastructure-as-code in Python, prefer the CDKTF — cdktf with cdktf get to generate provider bindings
• Manage state via the Terraform Cloud API workspaces and state-versions endpoints
• Use the tfe Terraform provider if you need to manage Terraform Cloud resources in HCL
• Authenticate with a TFE_TOKEN environment variable or explicit TFEConfig(token=...)
• Use Pulumi (see coding-pulumi skill) as an alternative when you prefer native IaC over Terraform wrappers

When to Use

Use this skill when:

• Automating workspace creation, runs, and state management in HCP Terraform or Terraform Enterprise
• Writing Python code that provisions infrastructure using Terraform providers (via CDKTF)
• Building CI/CD pipelines that trigger Terraform plans and applies
• Implementing policy-as-code with Sentinel or OPA policies applied to Terraform runs
• Managing Terraform state versions, outputs, and variables programmatically
• Extending Terraform with custom providers using the Terraform Plugin Framework

When NOT to Use

Avoid this skill for:

• Writing declarative HCL (use standard Terraform/OpenTofu CLI workflows instead)
• Low-level cloud API calls (use provider-specific SDKs like boto3 for AWS)
• Docker container management (use coding-docker-api)

Core Workflow

1. Authenticate to Terraform Cloud/Enterprise — Create a TFEClient with a TFEConfig containing the address and API token. Checkpoint: Call client.organizations.list() to verify connectivity and permissions.

2. Create or Select Workspace — Use client.workspaces.create() for new workspaces or client.workspaces.list() to find existing ones. Checkpoint: Verify workspace ID and that the execution mode matches your intent (local vs. remote).

3. Set Variables and Configuration — Use client.variables.create() for terraform variables and environment variables. Checkpoint: Confirm variable values are correct (mark sensitive variables as sensitive).

4. Trigger a Run — Create a new run via client.runs.create() with a configuration version. Monitor the run through its state machine: pending → planning → applying → applied. Checkpoint: Poll run.status and handle planning errors or apply failures.

5. Retrieve Outputs and State — Use client.state_versions.current() to get the current state and client.run.outputs() to extract terraform output values. Checkpoint: Validate outputs match expected schema.

Implementation Patterns

Pattern 1: CDKTF Python — Create AWS S3 Bucket

from constructs import Construct
from cdktf import App, TerraformStack, TerraformOutput
from cdktf_cdktf_provider_aws.s3_bucket import S3Bucket
from cdktf_cdktf_provider_aws.provider import AwsProvider


class S3BucketStack(TerraformStack):
    """Define an S3 bucket using CDKTF Python bindings."""

    def __init__(self, scope: Construct, id: str, bucket_name: str):
        super().__init__(scope, id)

        AwsProvider(self, "aws", region="us-east-1")

        bucket = S3Bucket(
            self,
            "my-bucket",
            bucket=bucket_name,
            versioning={
                "enabled": True,
            },
            tags={
                "Name": bucket_name,
                "ManagedBy": "CDKTF",
            },
        )

        TerraformOutput(self, "bucket_arn", value=bucket.arn)
        TerraformOutput(self, "bucket_id", value=bucket.id)


def deploy_s3_bucket(bucket_name: str) -> None:
    """Deploy an S3 bucket programmatically using CDKTF.

    Requires: pip install cdktf cdktf-cdktf-provider-aws

    Run with: python app.py && cdktf deploy
    """
    app = App()
    S3BucketStack(app, "s3-bucket-stack", bucket_name)
    app.synth()

Pattern 2: HCP Terraform Workspace Management (pyTFE)

import os
import time
from pytfe import TFEClient, TFEConfig
from pytfe.exceptions import TFEException


def create_workspace_and_run(
    organization: str,
    workspace_name: str,
    working_directory: str = "",
    variables: dict | None = None,
    auto_apply: bool = True,
    wait_for_completion: bool = True,
) -> dict:
    """Create a Terraform Cloud workspace and trigger a run.

    Args:
        organization: HCP Terraform organization name.
        workspace_name: Name for the new workspace.
        working_directory: Relative path to terraform config.
        variables: Terraform variables to set.
        auto_apply: Whether to auto-apply on plan success.
        wait_for_completion: Wait for the run to finish.

    Returns:
        Dict with workspace ID, run ID, and run status.

    Raises:
        TFEException: If API operations fail.
        TimeoutError: If the run doesn't complete.
    """
    config = TFEConfig(
        address=os.environ.get("TFE_ADDRESS", "https://app.terraform.io"),
        token=os.environ["TFE_TOKEN"],
    )
    client = TFEClient(config)

    # Create or find the workspace
    try:
        workspace = client.workspaces.create(
            organization=organization,
            name=workspace_name,
            auto_apply=auto_apply,
            working_directory=working_directory or None,
        )
    except TFEException as exc:
        # Workspace might already exist
        if "already exists" in str(exc):
            for ws in client.workspaces.list(organization).items:
                if ws.name == workspace_name:
                    workspace = ws
                    break
        else:
            raise

    # Set variables
    for key, value in (variables or {}).items():
        client.variables.create(
            workspace_id=workspace.id,
            key=key,
            value=str(value),
            category="terraform",
        )

    # Create a new configuration version (upload source)
    cv = client.configuration_versions.create(
        workspace_id=workspace.id,
        auto_queue_runs=True,
    )

    # Trigger the run
    run = client.runs.create(workspace_id=workspace.id)

    if wait_for_completion:
        _wait_for_run(client, run.id)

    return {
        "workspace_id": workspace.id,
        "workspace_name": workspace.name,
        "run_id": run.id,
        "status": run.status,
    }


def _wait_for_run(client: TFEClient, run_id: str, timeout: int = 600) -> None:
    """Poll a Terraform Cloud run until it completes or errors."""
    terminal_states = {"applied", "planned_and_finished", "errored", "canceled"}
    start = time.monotonic()

    while time.monotonic() - start < timeout:
        run = client.runs.read(run_id)
        if run.status in terminal_states:
            if run.status == "errored":
                raise RuntimeError(
                    f"Terraform run {run_id} failed with status: {run.status}"
                )
            return
        if run.status == "planning":
            _check_run_errors(run)
        time.sleep(5)

    raise TimeoutError(f"Terraform run {run_id} did not complete in {timeout}s.")

Pattern 3: Terraform Cloud State Version Retrieval

import os
from pytfe import TFEClient, TFEConfig
from pytfe.exceptions import TFEException


def get_workspace_outputs(
    organization: str,
    workspace_name: str,
) -> dict:
    """Retrieve Terraform outputs from a Cloud workspace's current state.

    Args:
        organization: HCP Terraform organization.
        workspace_name: Workspace name.

    Returns:
        Dict of output names to their values.

    Raises:
        RuntimeError: If workspace or state is not found.
    """
    config = TFEConfig(
        address=os.environ.get("TFE_ADDRESS", "https://app.terraform.io"),
        token=os.environ["TFE_TOKEN"],
    )
    client = TFEClient(config)

    # Find the workspace
    workspace = None
    for ws in client.workspaces.list(organization).items:
        if ws.name == workspace_name:
            workspace = ws
            break

    if workspace is None:
        raise RuntimeError(
            f"Workspace '{workspace_name}' not found in organization '{organization}'."
        )

    # Get current state version
    try:
        state_version = client.state_versions.current(workspace_id=workspace.id)
    except TFEException as exc:
        raise RuntimeError(
            f"Failed to retrieve current state for '{workspace_name}': {exc}"
        ) from exc

    # Parse and return outputs
    outputs: dict = {}
    if state_version and state_version.outputs:
        for output_name, output_data in state_version.outputs.items():
            outputs[output_name] = output_data.get("value")

    return outputs

BAD vs GOOD: Terraform API Error Handling

# ❌ BAD — silent pass, no state awareness
def get_workspace_bad(client, org):
    try:
        ws = client.workspaces.create(organization=org, name="my-ws")
        return ws
    except Exception:
        pass  # Ignores "already exists" and other failures

# ✅ GOOD — idempotent create-or-find with explicit error handling
def get_or_create_workspace(
    client: TFEClient,
    organization: str,
    name: str,
) -> dict:
    """Retrieve or create a Terraform Cloud workspace.

    Args:
        client: Authenticated TFEClient instance.
        organization: Organization name.
        name: Workspace name.

    Returns:
        Workspace object.

    Raises:
        RuntimeError: If creation fails for reasons other than conflict.
    """
    try:
        workspace = client.workspaces.create(organization=organization, name=name)
        return {"id": workspace.id, "name": workspace.name, "created": True}
    except TFEException as exc:
        if "already exists" in str(exc):
            for ws in client.workspaces.list(organization).items:
                if ws.name == name:
                    return {"id": ws.id, "name": ws.name, "created": False}
        raise RuntimeError(
            f"Failed to create workspace '{name}': {exc}"
        ) from exc

MUST DO

• Use environment variables (TFE_TOKEN, TFE_ADDRESS) for authentication — never hardcode tokens
• Implement idempotent create-or-find patterns for workspaces and resources
• Use category="terraform" for Terraform variables and category="env" for environment variables
• Poll run status with a timeout and handle all terminal states (applied, errored, canceled)
• Set auto_apply=True for workspaces where manual approval is not needed in CI/CD

MUST NOT DO

• Never store Terraform state management credentials in code — they provide full access to infrastructure
• Do not create workspaces without setting the execution mode explicitly
• Avoid hardcoding organization names — read them from environment configuration
• Never ignore variables that should be sensitive=true (e.g., database passwords, API keys)
• Do not assume workspaces are deletable — check for attached resources first

Constraints

MUST DO

• Validate all inputs at function boundaries before processing — guard clauses should fail early with descriptive errors
• Implement proper error handling that distinguishes between recoverable and unrecoverable failures
• Add comprehensive logging with structured context (correlation IDs, operation names, timing) for debugging and monitoring
• Write unit tests covering normal operations, edge cases, and error conditions before integrating the component

MUST NOT DO

• Do not silently swallow exceptions — always log or propagate errors with meaningful context
• Avoid unbounded resource allocation without limits (connection pools, memory buffers, thread counts)
• Never use hardcoded credentials, API keys, or secrets in source code
• Do not bypass input validation for perceived performance gains

Live References

• pyTFE (HCP Terraform Python Client) Documentation
• CDK for Terraform (CDKTF) Python API Reference
• HCP Terraform API Documentation
• Terraform Plugin Framework Documentation
• OpenTofu Documentation
• Terraform Registry — Providers
• Terraform Cloud State Versions API

Related Skills