feat(run-tasks-integration): add run tasks integration callback support

examples/run_tasks_integration.py

@@ -0,0 +1,208 @@
+"""
+Terraform Cloud/Enterprise Run Tasks Integration Example
+
+This example demonstrates how to use the python-tfe SDK to build a run task server
+that receives task requests from TFC/TFE and sends results back via the callback API.
+
+IMPORTANT: This example uses Flask as a simple HTTP server for demonstration purposes.
+You can use any web framework (FastAPI, Django, etc.) or even the built-in http.server.
+The key components are:
+1. Receiving POST requests with run task payloads
+2. Using TFEClient.run_tasks_integration.callback() to send results back
+
+Prerequisites:
+    - Install Flask (for this example only): pip install flask
+    - Expose your server publicly using ngrok, cloudflare tunnel, or similar
+    - Create a run task in TFC/TFE pointing to your public URL endpoint
+    - Attach the run task to a workspace
+
+Usage:
+    python examples/run_tasks_integration.py
+
+Then expose with ngrok:
+    ngrok http 5000
+
+API Documentation:
+    https://developer.hashicorp.com/terraform/enterprise/api-docs/run-tasks/run-tasks-integration
+"""
+
+from __future__ import annotations
+
+import os
+
+try:
+    from flask import Flask, request, jsonify
+except ImportError:
+    print("Error: Flask is required for this example")
+    print("Install it with: pip install flask")
+    exit(1)
+
+from pytfe import TFEClient, TFEConfig
+from pytfe.models import RunTaskRequest, RunTaskRequestCapabilities
+from pytfe.resources.run_tasks_integration import (
+    RunTasksIntegration,
+    TaskResultCallbackOptions,
+    TaskResultOutcome,
+    TaskResultStatus,
+    TaskResultTag,
+)
+
+app = Flask(__name__)
+
+# Initialize TFE client for callback functionality
+# Note: The callback uses the access_token from the run task request,
+# NOT your regular TFE API token
+config = TFEConfig()
+client = TFEClient(config)
+
+
+@app.route('/run-task', methods=['POST'])
+def handle_run_task():
+    """Handle incoming run task request from TFC/TFE."""
+    try:
+        # Parse the incoming request
+        run_task_request = RunTaskRequest(**request.json)
+
+        print(f"Received run task request:")
+        print(f"  Organization: {run_task_request.organization_name}")
+        print(f"  Workspace: {run_task_request.workspace_name}")
+        print(f"  Run ID: {run_task_request.run_id}")
+        print(f"  Stage: {run_task_request.stage}")
+        print(f"  Enforcement Level: {run_task_request.task_result_enforcement_level}")
+
+        # Extract the callback information
+        callback_url = run_task_request.task_result_callback_url
+        access_token = run_task_request.access_token
+
+        # YOUR CUSTOM LOGIC HERE
+        # This is where you would perform your actual run task checks
+        # For example:
+        # - Download and analyze the plan JSON
+        # - Check for policy violations
+        # - Validate resource configurations
+        # - Run security scans
+        # - Check cost estimates
+
+        # Example: Simple check based on workspace name
+        if "prod" in run_task_request.workspace_name.lower():
+            # Production workspace - run strict checks
+            result = perform_strict_checks(run_task_request)
+        else:
+            # Non-production - run basic checks
+            result = perform_basic_checks(run_task_request)
+
+        # Send the callback to TFC/TFE
+        callback_options = TaskResultCallbackOptions(
+            status=result["status"],
+            message=result["message"],
+            url=result.get("url"),
+            outcomes=result.get("outcomes", []),
+        )
+
+        client.run_tasks_integration.callback(
+            callback_url=callback_url,
+            access_token=access_token,
+            options=callback_options,
+        )
+
+        print(f"Successfully sent callback with status: {result['status']}")
+
+        # Return 200 OK to TFC/TFE
+        return jsonify({"status": "accepted"}), 200
+
+    except Exception as e:
+        print(f"Error processing run task: {e}")
+
+        # Even if processing fails, try to send a failure callback
+        try:
+            if 'callback_url' in locals() and 'access_token' in locals():
+                error_options = TaskResultCallbackOptions(
+                    status=TaskResultStatus.FAILED,
+                    message=f"Run task processing error: {str(e)}",
+                )
+                client.run_tasks_integration.callback(
+                    callback_url=callback_url,
+                    access_token=access_token,
+                    options=error_options,
+                )
+        except Exception as callback_error:
+            print(f"Failed to send error callback: {callback_error}")
+
+        return jsonify({"error": str(e)}), 500
+
+
+def perform_strict_checks(run_task_request: RunTaskRequest) -> dict:
+    """Perform strict checks for production workspaces.
+
+    This is a placeholder for your actual check logic.
+    """
+    # Example: Always pass for demo purposes
+    # In real implementation, you would:
+    # - Download the configuration or plan
+    # - Analyze it for compliance/security
+    # - Generate detailed outcomes
+
+    outcomes = [
+        TaskResultOutcome(
+            outcome_id="SECURITY-001",
+            description="Security check passed",
+            body="All security requirements met for production deployment.",
+            tags={
+                "Category": [TaskResultTag(label="Security")],
+                "Severity": [TaskResultTag(label="Info", level="info")],
+            },
+        ),
+        TaskResultOutcome(
+            outcome_id="COMPLIANCE-001",
+            description="Compliance check passed",
+            body="Configuration meets all compliance requirements.",
+            tags={
+                "Category": [TaskResultTag(label="Compliance")],
+                "Severity": [TaskResultTag(label="Info", level="info")],
+            },
+        ),
+    ]
+
+    return {
+        "status": TaskResultStatus.PASSED,
+        "message": "All production checks passed",
+        "url": "https://your-dashboard.example.com/results/123",
+        "outcomes": outcomes,
+    }
+
+
+def perform_basic_checks(run_task_request: RunTaskRequest) -> dict:
+    """Perform basic checks for non-production workspaces.
+
+    This is a placeholder for your actual check logic.
+    """
+    # Example: Simple validation
+    outcomes = [
+        TaskResultOutcome(
+            outcome_id="BASIC-001",
+            description="Basic validation passed",
+            body="Configuration syntax is valid.",
+            tags={
+                "Category": [TaskResultTag(label="Validation")],
+            },
+        ),
+    ]
+
+    return {
+        "status": TaskResultStatus.PASSED,
+        "message": "Basic checks completed successfully",
+        "outcomes": outcomes,
+    }
+
+
+@app.route('/health', methods=['GET'])
+def health_check():
+    """Health check endpoint."""
+    return jsonify({"status": "healthy"}), 200
+
+
+if __name__ == '__main__':
+    print("Starting Run Task server on http://localhost:5000")
+    print("Make sure to expose this with ngrok or similar for TFC/TFE to reach it")
+    print("Example: ngrok http 5000")
+    app.run(host='0.0.0.0', port=5000, debug=True)

src/pytfe/client.py

@@ -25,6 +25,7 @@
 from .resources.run import Runs
 from .resources.run_event import RunEvents
 from .resources.run_task import RunTasks
+from .resources.run_tasks_integration import RunTasksIntegration
 from .resources.run_trigger import RunTriggers
 from .resources.ssh_keys import SSHKeys
 from .resources.state_version_outputs import StateVersionOutputs
@@ -76,6 +77,7 @@ def __init__(self, config: TFEConfig | None = None):
         self.state_versions = StateVersions(self._transport)
         self.state_version_outputs = StateVersionOutputs(self._transport)
         self.run_tasks = RunTasks(self._transport)
+        self.run_tasks_integration = RunTasksIntegration(self._transport)
         self.run_triggers = RunTriggers(self._transport)
         self.runs = Runs(self._transport)
         self.query_runs = QueryRuns(self._transport)

src/pytfe/models/__init__.py

@@ -256,6 +256,10 @@
     Stage,
     TaskEnforcementLevel,
 )
+from .run_task_request import (
+    RunTaskRequest,
+    RunTaskRequestCapabilities,
+)
 from .run_trigger import (
     RunTrigger,
     RunTriggerCreateOptions,

src/pytfe/models/run_task_request.py

@@ -0,0 +1,118 @@
+"""Run Task Request models for python-tfe.
+
+This module contains the RunTaskRequest model which represents the payload
+that TFC/TFE sends to external run task servers.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class RunTaskRequestCapabilities(BaseModel):
+    """Capabilities that the caller supports."""
+
+    outcomes: bool = Field(
+        default=False,
+        description="Whether the run task server supports outcomes"
+    )
+
+
+class RunTaskRequest(BaseModel):
+    """Represents the payload that TFC/TFE sends to a run task's URL.
+
+    This is the incoming request that your external run task server receives
+    from Terraform Cloud/Enterprise when a run task is triggered.
+
+    API Documentation:
+        https://developer.hashicorp.com/terraform/enterprise/api-docs/run-tasks/run-tasks-integration#common-properties
+    """
+
+    access_token: str = Field(
+        description="Token to use for authentication when sending callback"
+    )
+    capabilities: RunTaskRequestCapabilities | None = Field(
+        default=None,
+        description="Capabilities that the caller supports"
+    )
+    configuration_version_download_url: str | None = Field(
+        default=None,
+        description="URL to download the configuration version"
+    )
+    configuration_version_id: str | None = Field(
+        default=None,
+        description="ID of the configuration version"
+    )
+    is_speculative: bool = Field(
+        description="Whether this is a speculative run"
+    )
+    organization_name: str = Field(
+        description="Name of the organization"
+    )
+    payload_version: int = Field(
+        description="Version of the payload format"
+    )
+    plan_json_api_url: str | None = Field(
+        default=None,
+        description="URL to access the plan JSON via API (post_plan, pre_apply, post_apply stages)"
+    )
+    run_app_url: str = Field(
+        description="URL to view the run in TFC/TFE UI"
+    )
+    run_created_at: datetime = Field(
+        description="Timestamp when the run was created"
+    )
+    run_created_by: str = Field(
+        description="Username of the user who created the run"
+    )
+    run_id: str = Field(
+        description="ID of the run"
+    )
+    run_message: str = Field(
+        description="Message associated with the run"
+    )
+    stage: str = Field(
+        description="Stage when the run task is executed (pre_plan, post_plan, pre_apply, post_apply)"
+    )
+    task_result_callback_url: str = Field(
+        description="URL to send the task result callback to"
+    )
+    task_result_enforcement_level: str = Field(
+        description="Enforcement level for the task result (advisory, mandatory)"
+    )
+    task_result_id: str = Field(
+        description="ID of the task result"
+    )
+    vcs_branch: str | None = Field(
+        default=None,
+        description="VCS branch name"
+    )
+    vcs_commit_url: str | None = Field(
+        default=None,
+        description="URL to the VCS commit"
+    )
+    vcs_pull_request_url: str | None = Field(
+        default=None,
+        description="URL to the VCS pull request"
+    )
+    vcs_repo_url: str | None = Field(
+        default=None,
+        description="URL to the VCS repository"
+    )
+    workspace_app_url: str = Field(
+        description="URL to view the workspace in TFC/TFE UI"
+    )
+    workspace_id: str = Field(
+        description="ID of the workspace"
+    )
+    workspace_name: str = Field(
+        description="Name of the workspace"
+    )
+    workspace_working_directory: str | None = Field(
+        default=None,
+        description="Working directory for the workspace"
+    )
+
+    model_config = ConfigDict(populate_by_name=True)