[Integrations][Github] GitHub ocean execution agent support

.vscode/launch.json

@@ -293,6 +293,17 @@
       "request": "launch",
       "type": "debugpy"
     },
+    {
+      "console": "integratedTerminal",
+      "cwd": "${workspaceFolder}/integrations/github",
+      "envFile": "${workspaceFolder}/integrations/github/.env",
+      "justMyCode": true,
+      "name": "Run Github integration",
+      "program": "${workspaceFolder}/integrations/github/debug.py",
+      "python": "${workspaceFolder}/integrations/github/.venv/bin/python",
+      "request": "launch",
+      "type": "debugpy"
+    },
     {
       "console": "integratedTerminal",
       "cwd": "${workspaceFolder}/integrations/http-server",

integrations/github/.port/spec.yaml

@@ -65,3 +65,24 @@ saas:
       integrationSpec:
         githubAppId: .oauthData.additionalData.appId
         githubOrganization: .oauthData.additionalData.githubOrganization
+executionAgent:
+  enabled: true
+  actions:
+  - name: dispatch_github_workflow
+    description: Dispatch a GitHub workflow
+    inputs:
+      - name: repo
+        type: string
+        description: The name of the GitHub repository that contains the target workflow
+        required: true
+      - name: workflow
+        type: string
+        description: The name of the file (under /.github/workflows/) that defines the target workflow, for example - deploy.yml
+        required: true
+      - name: workflowInputs
+        type: jqObject
+        description: 'The following values will be sent upon the action invocation. You can use JQ to reference the action trigger data by writing double brackets, for example: {{ .trigger.by.user.email }}. You can reference secrets using {{ .secrets["secret-name"] }}. <a href="https://docs.port.io/actions-and-automations/create-self-service-experiences/setup-the-backend/#define-the-actions-payload" target="_blank">Learn more</a>'
+      - name: reportWorkflowStatus
+        type: boolean
+        description: Whether to report the workflow status
+        default: true

integrations/github/CHANGELOG.md

@@ -6,6 +6,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 <!-- towncrier release notes start -->
+## 2.1.0-beta (2025-10-16)
+
+
+### Improvements
+
+- Added support for running github workflows as part of Port actions
 
 ## 2.0.1-beta (2025-10-15)
 

integrations/github/github/actions/abstract_github_executor.py

@@ -0,0 +1,24 @@
+from integrations.github.github.clients.client_factory import create_github_client
+from port_ocean.core.handlers.actions.abstract_executor import AbstractExecutor
+
+
+MIN_REMAINING_RATE_LIMIT_FOR_EXECUTE_WORKFLOW = 20
+
+
+class AbstractGithubExecutor(AbstractExecutor):
+    def __init__(self):
+        self.rest_client = create_github_client()
+
+    async def is_close_to_rate_limit(self) -> bool:
+        info = self.rest_client.get_rate_limit_status()
+        if not info:
+            return False
+
+        return info.remaining < MIN_REMAINING_RATE_LIMIT_FOR_EXECUTE_WORKFLOW
+
+    async def get_remaining_seconds_until_rate_limit(self) -> float:
+        info = self.rest_client.get_rate_limit_status()
+        if not info:
+            return 0
+
+        return info.seconds_until_reset

integrations/github/github/actions/dispatch_workflow_executor.py

@@ -0,0 +1,270 @@
+import asyncio
+from datetime import datetime, timezone
+import json
+
+import httpx
+from loguru import logger
+from integrations.github.github.actions.utils import build_external_id
+from integrations.github.github.context.auth import (
+    get_authenticated_user,
+)
+from integrations.github.github.core.exporters.repository_exporter import (
+    RestRepositoryExporter,
+)
+from integrations.github.github.core.options import SingleRepositoryOptions
+from integrations.github.github.webhook.registry import (
+    WEBHOOK_PATH as DISPATCH_WEBHOOK_PATH,
+)
+from port_ocean.context.ocean import ocean
+from port_ocean.core.handlers.port_app_config.models import ResourceConfig
+from port_ocean.core.handlers.queue.group_queue import MaybeStr
+from port_ocean.core.handlers.webhook.abstract_webhook_processor import (
+    WebhookProcessorType,
+)
+from port_ocean.core.handlers.webhook.webhook_event import (
+    EventPayload,
+    WebhookEvent,
+    WebhookEventRawResults,
+)
+from integrations.github.github.webhook.webhook_processors.base_workflow_run_webhook_processor import (
+    BaseWorkflowRunWebhookProcessor,
+)
+from port_ocean.core.models import (
+    ActionRun,
+    IntegrationActionInvocationPayload,
+    RunStatus,
+)
+from integrations.github.github.actions.abstract_github_executor import (
+    AbstractGithubExecutor,
+)
+
+
+MAX_WORKFLOW_POLL_ATTEMPTS = 10
+WORKFLOW_POLL_DELAY_SECONDS = 2
+
+
+class DispatchWorkflowExecutor(AbstractGithubExecutor):
+    """
+    Executor for dispatching GitHub workflow runs and tracking their execution.
+
+    This executor implements the Port action for triggering GitHub Actions workflows.
+    It supports:
+    - Dispatching workflows with custom inputs
+    - Tracking workflow execution status
+    - Reporting workflow completion back to Port
+    - Rate limit handling for GitHub API
+    - Webhook processing for async status updates
+
+    The executor uses workflow names as partition keys to ensure sequential
+    execution of the same workflow, which is necessary for proper run tracking.
+    It identifies workflow runs by finding the one closest to the trigger time
+    and recording its ID.
+
+    Attributes:
+        ACTION_NAME (str): The name of this action in Port's spec ("dispatch_workflow")
+        PARTITION_KEY (str): The key for partitioning runs ("workflow")
+        WEBHOOK_PROCESSOR_CLASS (Type[AbstractWebhookProcessor]): Processor for workflow_run events
+        WEBHOOK_PATH (str): Path for receiving GitHub webhook events
+        _default_ref_cache (dict[str, str]): Cache of repository default branch names
+
+    Example Usage in Port:
+        ```yaml
+        actions:
+          dispatch_workflow:
+            displayName: Trigger Workflow
+            trigger: PORT
+            inputs:
+              repo:
+                type: string
+                description: Repository name
+              workflow:
+                type: string
+                description: Workflow file name or ID
+              workflowInputs:
+                type: object
+                description: Optional workflow inputs
+        ```
+
+    Example API Usage:
+        ```python
+        executor = DispatchWorkflowExecutor()
+        await executor.execute(ActionRun(
+            payload=IntegrationActionInvocationPayload(
+                actionType="dispatch_workflow",
+                integrationActionExecutionProperties={
+                    "repo": "my-repo",
+                    "workflow": "deploy.yml",
+                    "workflowInputs": {"environment": "prod"}
+                }
+            )
+        ))
+        ```
+    """
+
+    ACTION_NAME = "dispatch_workflow"
+
+    """
+    We use the workflow name as the partition key because we track workflow executions
+    by locating the workflow run closest to the trigger time and record its ID.
+    Triggering the same workflow concurrently would prevent us from uniquely tracking each instance.
+    """
+
+    class DispatchWorkflowWebhookProcessor(BaseWorkflowRunWebhookProcessor):
+        """
+        Webhook processor for handling GitHub workflow run completion events.
+
+        This processor is responsible for:
+        1. Filtering workflow_run events to only process completed runs
+        2. Verifying that the run was triggered by the authenticated user
+        3. Updating the Port action run status based on the workflow conclusion
+        4. Handling the mapping between GitHub run IDs and Port run IDs
+
+        The processor only handles events where:
+        - The event type is workflow_run
+        - The workflow run status is "completed"
+        - The actor matches the authenticated GitHub user
+        - The run has a matching Port action run ID
+
+        Attributes:
+            Inherits all attributes from BaseWorkflowRunWebhookProcessor
+        """
+
+        @classmethod
+        def get_processor_type(cls) -> WebhookProcessorType:
+            return WebhookProcessorType.ACTION
+
+        async def _should_process_event(self, event: WebhookEvent) -> bool:
+            """
+            Determine if this webhook event should be processed.
+            """
+            workflow_run = event.payload["workflow_run"]
+            authenticated_user = await get_authenticated_user()
+            should_process = (
+                await super()._should_process_event(event)
+                and workflow_run["status"] == "completed"
+                and workflow_run["actor"]["login"] == authenticated_user.login
+            )
+            return should_process
+
+        async def handle_event(
+            self, payload: EventPayload, resource_config: ResourceConfig
+        ) -> WebhookEventRawResults:
+            """
+            Handle a workflow run completion webhook event.
+            """
+            workflow_run = payload["workflow_run"]
+
+            external_id = build_external_id(workflow_run)
+            action_run: ActionRun[IntegrationActionInvocationPayload] | None = (
+                await ocean.port_client.get_run_by_external_id(external_id)
+            )
+
+            if (
+                action_run
+                and action_run.status == RunStatus.IN_PROGRESS
+                and action_run.payload.integrationActionExecutionProperties.get(
+                    "reportWorkflowStatus", False
+                )
+            ):
+                status = (
+                    RunStatus.SUCCESS
+                    if workflow_run["conclusion"] in ["success", "skipped", "neutral"]
+                    else RunStatus.FAILURE
+                )
+                await ocean.port_client.patch_run(action_run.id, {"status": status})
+
+            return WebhookEventRawResults(
+                updated_raw_results=[], deleted_raw_results=[]
+            )
+
+    WEBHOOK_PROCESSOR_CLASS = DispatchWorkflowWebhookProcessor
+    WEBHOOK_PATH = DISPATCH_WEBHOOK_PATH
+    _default_ref_cache: dict[str, str] = {}
+
+    async def _get_partition_key(
+        self, run: ActionRun[IntegrationActionInvocationPayload]
+    ) -> MaybeStr:
+        """
+        Get the workflow name as the partition key.
+        """
+        return run.payload.integrationActionExecutionProperties.get("workflow")
+
+    async def _get_default_ref(self, repo_name: str) -> str:
+        """
+        Get the default branch name for a repository, using cache when available.
+        """
+        if repo_name in self._default_ref_cache:
+            return self._default_ref_cache[repo_name]
+
+        repoExporter = RestRepositoryExporter(self.rest_client)
+        repo = await repoExporter.get_resource(SingleRepositoryOptions(name=repo_name))
+        if not repo.get("default_branch"):
+            raise Exception(f"Failed to get repository data for {repo_name}")
+
+        self._default_ref_cache[repo_name] = repo["default_branch"]
+        return self._default_ref_cache[repo_name]
+
+    async def execute(self, run: ActionRun[IntegrationActionInvocationPayload]) -> None:
+        """
+        Execute a workflow dispatch action by triggering a GitHub Actions workflow.
+        """
+        repo = run.payload.integrationActionExecutionProperties.get("repo")
+        workflow = run.payload.integrationActionExecutionProperties.get("workflow")
+        inputs = run.payload.integrationActionExecutionProperties.get(
+            "workflowInputs", {}
+        )
+
+        if not (repo and workflow):
+            raise ValueError("repo and workflow are required")
+
+        ref = await self._get_default_ref(repo)
+        try:
+            isoDate = datetime.now(timezone.utc).isoformat()
+            await self.rest_client.make_request(
+                f"{self.rest_client.base_url}/repos/{self.rest_client.organization}/{repo}/actions/workflows/{workflow}/dispatches",
+                method="POST",
+                json_data={
+                    "ref": ref,
+                    "inputs": inputs,
+                },
+                ignore_default_errors=False,
+            )
+
+            # Get the workflow run id
+            workflow_runs = []
+            attempts_made = 0
+            while (
+                len(workflow_runs) == 0 and attempts_made < MAX_WORKFLOW_POLL_ATTEMPTS
+            ):
+                authenticated_user = await get_authenticated_user()
+                response = await self.rest_client.send_api_request(
+                    f"{self.rest_client.base_url}/repos/{self.rest_client.organization}/{repo}/actions/runs",
+                    params={
+                        "actor": authenticated_user.login,
+                        "event": "workflow_dispatch",
+                        "created": f">{isoDate}",
+                        "exclude_pull_requests": True,
+                        "branch": ref,
+                    },
+                    method="GET",
+                    ignore_default_errors=False,
+                )
+                workflow_runs = response.get("workflow_runs", [])
+                if len(workflow_runs) == 0:
+                    logger.warning(
+                        f"Couldn't find the triggered workflow run, waiting for {WORKFLOW_POLL_DELAY_SECONDS} seconds",
+                        attempts_made=attempts_made,
+                    )
+                    await asyncio.sleep(WORKFLOW_POLL_DELAY_SECONDS)
+                    attempts_made += 1
+
+            if len(workflow_runs) == 0:
+                raise Exception("No workflow runs found")
+
+            external_id = build_external_id(workflow_runs[0])
+            await ocean.port_client.patch_run(run.id, {"external_run_id": external_id})
+        except Exception as e:
+            error_message = str(e)
+            if isinstance(e, httpx.HTTPStatusError):
+                error_message = json.loads(e.response.text).get("message", str(e))
+            raise Exception(f"Error dispatching workflow: {error_message}")

integrations/github/github/actions/registry.py

@@ -0,0 +1,9 @@
+from port_ocean.context.ocean import ocean
+from integrations.github.github.actions.dispatch_workflow_executor import (
+    DispatchWorkflowExecutor,
+)
+
+
+def register_actions_executors():
+    """Register all actions executors."""
+    ocean.register_action_executor(DispatchWorkflowExecutor())

integrations/github/github/actions/utils.py

@@ -0,0 +1,5 @@
+from typing import Any
+
+
+def build_external_id(workflow_run: dict[str, Any]) -> str:
+    return f'gh_{workflow_run["repository"]["owner"]["id"]}_{workflow_run["repository"]["id"]}_{workflow_run["id"]}'