Run task request & Workspace Run_task

examples/workspace_run_task.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""
+Workspace Run Task Operations Example
+
+Demonstrates workspace run task operations:
+1. create() - Attach a run task to a workspace
+2. list() - List all workspace run tasks for a workspace
+3. read() - Read a workspace run task by ID
+4. update() - Update enforcement/stage settings
+5. delete() - Delete a workspace run task
+
+Prerequisites:
+- Set TFE_TOKEN and TFE_ADDRESS environment variables
+- Use valid workspace and run task IDs
+"""
+
+import argparse
+
+from pytfe import TFEClient, TFEConfig
+from pytfe.models import (
+    WorkspaceRunTaskCreateOptions,
+    WorkspaceRunTaskUpdateOptions,
+)
+
+
+def _find_matching_workspace_run_task(items, run_task_id: str):
+    for item in items:
+        if item.run_task and item.run_task.id == run_task_id:
+            return item
+
+    if len(items) == 1:
+        return items[0]
+
+    return None
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Workspace run task operations demo for python-tfe SDK"
+    )
+    parser.add_argument(
+        "--workspace-id",
+        required=True,
+        help="Workspace ID (example: ws-abc123)",
+    )
+    parser.add_argument(
+        "--run-task-id",
+        required=True,
+        help="Run task ID (example: task-abc123)",
+    )
+    parser.add_argument(
+        "--delete-existing",
+        action="store_true",
+        help="Allow delete() to remove a workspace run task that existed before this example ran",
+    )
+    args = parser.parse_args()
+
+    client = TFEClient(TFEConfig.from_env())
+
+    workspace_id = args.workspace_id
+    run_task_id = args.run_task_id
+
+    if workspace_id == "ws-xxxxxxxx" or run_task_id == "task-xxxxxxxx":
+        print("Please provide real IDs for --workspace-id and --run-task-id")
+        return
+
+    print("=" * 80)
+    print("WORKSPACE RUN TASK OPERATIONS")
+    print("=" * 80)
+
+    workspace_task = None
+    created_in_run = False
+
+    print("\n1. create()")
+    try:
+        create_options = WorkspaceRunTaskCreateOptions(
+            enforcement_level="advisory",
+            run_task={"id": run_task_id},
+            stages=["post_plan"],
+        )
+        workspace_task = client.workspace_run_tasks.create(workspace_id, create_options)
+        created_in_run = True
+        print(f"Created workspace run task: {workspace_task.id}")
+    except Exception as exc:
+        print(f"Create result: {exc}")
+
+    print("\n2. list()")
+    items = []
+    try:
+        items = list(client.workspace_run_tasks.list(workspace_id))
+        print(f"Found {len(items)} workspace run task(s)")
+        for item in items:
+            run_task = item.run_task.id if item.run_task else None
+            print(
+                f"- {item.id} run_task={run_task} enforcement={item.enforcement_level} stages={item.stages}"
+            )
+
+        if workspace_task is None:
+            workspace_task = _find_matching_workspace_run_task(items, run_task_id)
+            if workspace_task is not None:
+                print(f"Using existing workspace run task: {workspace_task.id}")
+    except Exception as exc:
+        print(f"List failed: {exc}")
+
+    print("\n3. read()")
+    if workspace_task is None:
+        print("Read skipped: no workspace run task ID available")
+    else:
+        try:
+            workspace_task = client.workspace_run_tasks.read(
+                workspace_id, workspace_task.id
+            )
+            print(
+                f"Read workspace run task: {workspace_task.id} enforcement={workspace_task.enforcement_level} stages={workspace_task.stages}"
+            )
+        except Exception as exc:
+            print(f"Read failed: {exc}")
+
+    print("\n4. update()")
+    if workspace_task is None:
+        print("Update skipped: no workspace run task ID available")
+    else:
+        try:
+            update_options = WorkspaceRunTaskUpdateOptions(
+                # enforcement_level="mandatory",
+                stages=["post_plan", "pre_apply"],
+            )
+            workspace_task = client.workspace_run_tasks.update(
+                workspace_id, workspace_task.id, update_options
+            )
+            print(
+                f"Updated workspace run task: {workspace_task.id} enforcement={workspace_task.enforcement_level} stages={workspace_task.stages}"
+            )
+        except Exception as exc:
+            print(f"Update failed: {exc}")
+
+    print("\n5. delete()")
+    if workspace_task is None:
+        print("Delete skipped: no workspace run task ID available")
+    elif not created_in_run and not args.delete_existing:
+        print(
+            "Delete skipped: workspace run task existed before this example. "
+            "Re-run with --delete-existing to delete it."
+        )
+    else:
+        try:
+            client.workspace_run_tasks.delete(workspace_id, workspace_task.id)
+            print(f"Deleted workspace run task: {workspace_task.id}")
+        except Exception as exc:
+            print(f"Delete failed: {exc}")
+
+    print("=" * 80)
+    print("WORKSPACE RUN TASK OPERATIONS COMPLETED")
+    print("=" * 80)
+
+
+if __name__ == "__main__":
+    main()

src/pytfe/client.py

@@ -52,6 +52,7 @@
 from .resources.variable import Variables
 from .resources.variable_sets import VariableSets, VariableSetVariables
 from .resources.workspace_resources import WorkspaceResourcesService
+from .resources.workspace_run_task import WorkspaceRunTasks
 from .resources.workspaces import Workspaces
 
 
@@ -103,6 +104,7 @@ def __init__(self, config: TFEConfig | None = None):
         self.variable_set_variables = VariableSetVariables(self._transport)
         self.workspaces = Workspaces(self._transport)
         self.workspace_resources = WorkspaceResourcesService(self._transport)
+        self.workspace_run_tasks = WorkspaceRunTasks(self._transport)
         self.registry_modules = RegistryModules(self._transport)
         self.registry_providers = RegistryProviders(self._transport)
         self.registry_provider_versions = RegistryProviderVersions(self._transport)

src/pytfe/errors.py

@@ -338,6 +338,13 @@ def __init__(self, message: str = 'category must be "task"'):
         super().__init__(message)
 
 
+class InvalidWorkspaceRunTaskIDError(InvalidValues):
+    """Raised when an invalid workspace run task ID is provided."""
+
+    def __init__(self, message: str = "invalid value for workspace run task ID"):
+        super().__init__(message)
+
+
 # Run Trigger errors
 class RequiredRunTriggerListOpsError(RequiredFieldMissing):
     """Raised when required run trigger list options are missing."""

src/pytfe/models/__init__.py

@@ -328,6 +328,10 @@
 from .run_task_integration import (
     TaskResultStatus as TaskResultCallbackStatus,
 )
+from .run_task_request import (
+    RunTaskRequest,
+    RunTaskRequestCapabilitites,
+)
 from .run_trigger import (
     RunTrigger,
     RunTriggerCreateOptions,
@@ -456,6 +460,15 @@
     WorkspaceResource,
     WorkspaceResourceListOptions,
 )
+from .workspace_run_task import (
+    RunTaskReference,
+    WorkspaceRunTask,
+    WorkspaceRunTaskCreateOptions,
+    WorkspaceRunTaskEnforcementLevel,
+    WorkspaceRunTaskListOptions,
+    WorkspaceRunTaskStage,
+    WorkspaceRunTaskUpdateOptions,
+)
 
 # ── Public surface ────────────────────────────────────────────────────────────
 __all__ = [
@@ -684,6 +697,14 @@
     # Workspace Resources
     "WorkspaceResource",
     "WorkspaceResourceListOptions",
+    # Workspace Run Tasks
+    "RunTaskReference",
+    "WorkspaceRunTask",
+    "WorkspaceRunTaskEnforcementLevel",
+    "WorkspaceRunTaskStage",
+    "WorkspaceRunTaskListOptions",
+    "WorkspaceRunTaskCreateOptions",
+    "WorkspaceRunTaskUpdateOptions",
     "RunQueue",
     "ReadRunQueueOptions",
     # Runs
@@ -728,6 +749,9 @@
     "RunTaskCreateOptions",
     "RunTaskUpdateOptions",
     "RunTaskReadOptions",
+    # Run Task Request
+    "RunTaskRequest",
+    "RunTaskRequestCapabilitites",
     # Task Result
     "TaskResult",
     "TaskResultEnforcementLevel",

src/pytfe/models/run_task_request.py

@@ -0,0 +1,119 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class RunTaskRequestCapabilitites(BaseModel):
+    """Defines the capabilities that the caller supports."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    outcomes: bool = Field(..., description="Whether the caller supports outcomes")
+
+
+class RunTaskRequest(BaseModel):
+    """Payload object that TFC/E sends to the Run Task's URL.
+
+    https://developer.hashicorp.com/terraform/enterprise/api-docs/run-tasks/run-tasks-integration#common-properties
+    """
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    access_token: str = Field(
+        ..., alias="access_token", description="The access token for the run task"
+    )
+    capabilitites: RunTaskRequestCapabilitites = Field(
+        default_factory=lambda: RunTaskRequestCapabilitites(outcomes=False),
+        alias="capabilitites",
+        description="The capabilities that the caller supports",
+    )
+    configuration_version_download_url: str | None = Field(
+        None,
+        alias="configuration_version_download_url",
+        description="The URL to download the configuration version",
+    )
+    configuration_version_id: str | None = Field(
+        None,
+        alias="configuration_version_id",
+        description="The ID of the configuration version",
+    )
+    is_speculative: bool = Field(
+        ..., alias="is_speculative", description="Whether the run is speculative"
+    )
+    organization_name: str = Field(
+        ..., alias="organization_name", description="The name of the organization"
+    )
+    payload_version: int = Field(
+        ..., alias="payload_version", description="The version of the payload format"
+    )
+    plan_json_api_url: str | None = Field(
+        None,
+        alias="plan_json_api_url",
+        description="URL to the plan JSON API (specific to post_plan, pre_apply or post_apply stage)",
+    )
+    run_app_url: str = Field(
+        ..., alias="run_app_url", description="The URL to the run in the TFC/E UI"
+    )
+    run_created_at: datetime = Field(
+        ..., alias="run_created_at", description="The time the run was created"
+    )
+    run_created_by: str = Field(
+        ..., alias="run_created_by", description="The user who created the run"
+    )
+    run_id: str = Field(..., alias="run_id", description="The ID of the run")
+    run_message: str = Field(
+        ..., alias="run_message", description="The message associated with the run"
+    )
+    stage: str = Field(..., alias="stage", description="The stage of the run task")
+    task_result_callback_url: str = Field(
+        ...,
+        alias="task_result_callback_url",
+        description="The URL to call with the task result",
+    )
+    task_result_enforcement_level: str = Field(
+        ...,
+        alias="task_result_enforcement_level",
+        description="The enforcement level of the task result",
+    )
+    task_result_id: str = Field(
+        ..., alias="task_result_id", description="The ID of the task result"
+    )
+    vcs_branch: str | None = Field(
+        None, alias="vcs_branch", description="The VCS branch associated with the run"
+    )
+    vcs_commit_url: str | None = Field(
+        None,
+        alias="vcs_commit_url",
+        description="The URL of the VCS commit associated with the run",
+    )
+    vcs_pull_request_url: str | None = Field(
+        None,
+        alias="vcs_pull_request_url",
+        description="The URL of the VCS pull request associated with the run",
+    )
+    vcs_repo_url: str | None = Field(
+        None,
+        alias="vcs_repo_url",
+        description="The URL of the VCS repository associated with the run",
+    )
+    workspace_app_url: str = Field(
+        ...,
+        alias="workspace_app_url",
+        description="The URL to the workspace in the TFC/E UI",
+    )
+    workspace_id: str = Field(
+        ..., alias="workspace_id", description="The ID of the workspace"
+    )
+    workspace_name: str = Field(
+        ..., alias="workspace_name", description="The name of the workspace"
+    )
+    workspace_working_directory: str | None = Field(
+        None,
+        alias="workspace_working_directory",
+        description="The working directory configured for the workspace",
+    )