Implement Run Task integration callback API support

examples/run_task_integration.py

@@ -0,0 +1,130 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
+"""Example Run Task callback integration.
+
+This example sends a callback result back to Terraform after a Run Task
+webhook is received.
+
+Required environment variables:
+
+- TFE_ADDRESS
+    Terraform address (for example: https://app.terraform.io)
+
+- TFE_TOKEN
+    Your Terraform API token used to initialize the SDK client.
+
+- TFE_CALLBACK_URL
+    The task_result_callback_url received in the Run Task webhook payload.
+
+- TFE_CALLBACK_TOKEN
+    The access_token received in the same webhook payload.
+    This token is used for the callback request and is different from
+    your regular Terraform API token.
+
+Local testing flow:
+
+1. Start the webhook server:
+
+       uvicorn examples.run_task_webhook_server:app --reload --port 8000
+
+2. Expose the server publicly:
+
+       ngrok http 8000
+
+3. Create a Run Task in Terraform Cloud / Enterprise using the ngrok URL.
+
+4. Attach the Run Task to a workspace and trigger a run.
+
+5. The webhook payload will include values similar to:
+
+       {
+         "task_result_callback_url": "https://app.terraform.io/...",
+         "access_token": "v1.xxxxx..."
+       }
+
+6. Export those values locally and run this example script,
+   or call client.run_task_integrations.callback(...) directly
+   inside your webhook handler.
+
+Example:
+
+    export TFE_ADDRESS=https://app.terraform.io
+    export TFE_TOKEN=<your-api-token>
+    export TFE_CALLBACK_URL=<from-webhook-payload>
+    export TFE_CALLBACK_TOKEN=<from-webhook-payload>
+
+    python examples/run_task_integration.py
+"""
+
+from __future__ import annotations
+
+import os
+
+from pytfe import TFEClient, TFEConfig
+from pytfe.models.run_task_integration import (
+    TaskResultCallbackRequestOptions,
+    TaskResultOutcome,
+    TaskResultStatus,
+    TaskResultTag,
+)
+
+
+def main() -> None:
+    callback_url = os.getenv("TFE_CALLBACK_URL")
+    access_token = os.getenv("TFE_CALLBACK_TOKEN")
+
+    if not callback_url or not access_token:
+        print("Missing TFE_CALLBACK_URL or TFE_CALLBACK_TOKEN")
+        return
+
+    # TFE_ADDRESS and TFE_TOKEN are loaded from the environment.
+    # The callback request itself uses the short-lived webhook token.
+    client = TFEClient(TFEConfig.from_env())
+
+    outcome = TaskResultOutcome(
+        description="Example outcome",
+        body="All checks passed successfully",
+        tags={"severity": [TaskResultTag(label="low", level="info")]},
+    )
+
+    # Example status values:
+    #
+    # - passed: marks the run task as successful
+    # - failed: fails the run task
+    # - running: reports progress before sending a final result
+    #
+    # Example: send an in-progress update
+    #
+    # options = TaskResultCallbackRequestOptions(
+    #     status=TaskResultStatus.running,
+    #     message="Security scan in progress",
+    # )
+    #
+    # Example: report a failure
+    #
+    # options = TaskResultCallbackRequestOptions(
+    #     status=TaskResultStatus.failed,
+    #     message="Found critical vulnerabilities",
+    # )
+
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed,
+        message="Run task completed successfully",
+        url="https://example.com/results",
+        outcomes=[outcome],
+    )
+
+    print(f"Sending callback to: {callback_url}")
+
+    client.run_task_integrations.callback(
+        callback_url=callback_url,
+        access_token=access_token,
+        options=options,
+    )
+
+    print("Run task callback sent successfully")
+
+
+if __name__ == "__main__":
+    main()

examples/run_task_webhook_server.py

@@ -0,0 +1,103 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
+"""Minimal FastAPI webhook server for Terraform Run Tasks.
+
+This example receives a Run Task webhook, extracts the callback URL
+and access token from the payload, and sends a callback result back
+to Terraform using the SDK.
+
+Setup:
+
+1. Install dependencies:
+
+       pip install fastapi uvicorn
+
+2. Configure environment variables:
+
+       export TFE_ADDRESS=https://app.terraform.io
+       export TFE_TOKEN=<your-api-token>
+
+3. Start the server from the repository root:
+
+       uvicorn examples.run_task_webhook_server:app --reload --port 8000
+
+4. Expose the server publicly with ngrok:
+
+       ngrok http 8000
+
+5. In Terraform Cloud / Enterprise:
+
+   - Create a Run Task using the ngrok URL
+   - Attach the Run Task to a workspace
+   - Trigger a Terraform run
+
+The webhook payload will include values like:
+
+       {
+         "task_result_callback_url": "...",
+         "access_token": "..."
+       }
+
+This example prints the payload locally and sends a successful
+callback response back to Terraform.
+"""
+
+from __future__ import annotations
+
+import json
+
+from fastapi import FastAPI, Request
+
+from pytfe import TFEClient, TFEConfig
+from pytfe.models.run_task_integration import (
+    TaskResultCallbackRequestOptions,
+    TaskResultStatus,
+)
+
+app = FastAPI()
+client = TFEClient(TFEConfig.from_env())
+
+
+@app.post("/")
+async def receive_webhook(request: Request) -> dict[str, bool]:
+    try:
+        payload = await request.json()
+    except Exception:
+        # Terraform verification requests may not include a JSON payload.
+        return {"ok": True}
+
+    print("\n=== FULL PAYLOAD ===")
+    print(json.dumps(payload, indent=2))
+
+    callback_url = payload.get("task_result_callback_url")
+    access_token = payload.get("access_token")
+
+    print("\n=== EXTRACTED VALUES ===")
+    print("callback_url:", callback_url)
+    print("access_token:", access_token)
+
+    if not callback_url or not access_token:
+        # Verification requests do not include callback information.
+        return {"ok": True}
+
+    options = TaskResultCallbackRequestOptions(
+        status=TaskResultStatus.passed,
+        message="Webhook received and processed",
+        url="https://github.com/hashicorp/python-tfe",
+    )
+
+    print(f"Sending callback to: {callback_url}")
+
+    try:
+        client.run_task_integrations.callback(
+            callback_url=callback_url,
+            access_token=access_token,
+            options=options,
+        )
+        print("Run task callback sent successfully")
+    except Exception as exc:
+        print(f"Callback failed: {exc!r}")
+        return {"ok": False}
+
+    return {"ok": True}

src/pytfe/client.py

@@ -33,6 +33,7 @@
 from .resources.run import Runs
 from .resources.run_event import RunEvents
 from .resources.run_task import RunTasks
+from .resources.run_task_integration import RunTaskIntegrations
 from .resources.run_trigger import RunTriggers
 from .resources.ssh_keys import SSHKeys
 from .resources.stack import Stacks
@@ -100,6 +101,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_task_integrations = RunTaskIntegrations(self._transport)
         self.run_triggers = RunTriggers(self._transport)
         self.runs = Runs(self._transport)
         self.query_runs = QueryRuns(self._transport)

src/pytfe/errors.py

@@ -58,6 +58,24 @@ class RequiredFieldMissing(TFEError): ...
 class ErrStateVersionUploadNotSupported(TFEError): ...
 
 
+class InvalidCallbackURLError(TFEError):
+    def __init__(self, message: str = "Invalid callback URL") -> None:
+        super().__init__(message)
+
+
+class InvalidAccessTokenError(TFEError):
+    def __init__(self, message: str = "Invalid access token") -> None:
+        super().__init__(message)
+
+
+class InvalidTaskResultsCallbackStatusError(TFEError):
+    def __init__(
+        self,
+        message: str = "Invalid task result callback status; must be one of: passed, failed, running",
+    ) -> None:
+        super().__init__(message)
+
+
 # Generic error constants
 ERR_UNAUTHORIZED = "unauthorized"
 ERR_RESOURCE_NOT_FOUND = "resource not found"

src/pytfe/models/__init__.py

@@ -296,6 +296,14 @@
     Stage,
     TaskEnforcementLevel,
 )
+from .run_task_integration import (
+    TaskResultCallbackRequestOptions,
+    TaskResultOutcome,
+    TaskResultTag,
+)
+from .run_task_integration import (
+    TaskResultStatus as TaskResultCallbackStatus,
+)
 from .run_trigger import (
     RunTrigger,
     RunTriggerCreateOptions,
@@ -654,6 +662,11 @@
     "RunTaskCreateOptions",
     "RunTaskUpdateOptions",
     "RunTaskReadOptions",
+    # Run task integration (callback)
+    "TaskResultCallbackRequestOptions",
+    "TaskResultCallbackStatus",
+    "TaskResultOutcome",
+    "TaskResultTag",
     # Run triggers
     "RunTrigger",
     "RunTriggerCreateOptions",