feat: add run action tool (#31)

Author: danielsinai

CHANGELOG.md

@@ -5,6 +5,15 @@ All notable changes to this project will be documented in this file.
 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).
 
+## [0.2.12] - 2025-06-09
+
+### Changed
+- Added `run_action` tool to run action
+- Added `get_actions` tool to list actions 
+- Added `track_action_run` tool to track action runs
+- Added `get_action` tool to get action by identifier
+- Fixed mypy errors
+
 ## [0.2.11] - 2025-06-04
 
 ### Changed

Dockerfile

@@ -41,4 +41,4 @@ RUN chmod +x /app/entrypoint.sh
 ENTRYPOINT ["/app/entrypoint.sh"]
 
 # The server will expect client_id and client_secret to be provided via environment variables:
-# PORT_CLIENT_ID and PORT_CLIENT_SECRET
+# PORT_CLIENT_ID and PORT_CLIENT_SECRET

poetry.lock

@@ -1,6 +1,6 @@
 [project]
 name = "mcp-server-port"
-version = "0.2.11"
+version = "0.2.12"
 authors = [
   { name = "Matan Grady", email = "matan.grady@getport.io" }
 ]
@@ -99,4 +99,4 @@ isort = "^5.12.0"
 pre-commit = "^3.5.0"
 build = "^1.0.3"
 twine = "^4.0.2"
-pyright = "^1.1.389"
+pyright = "^1.1.389"

pyproject.toml

@@ -1,9 +1,17 @@
 """Client package for Port.io API interactions."""
 
+from .actions import PortActionClient
 from .agent import PortAgentClient
 from .blueprints import PortBlueprintClient
 from .client import PortClient
 from .entities import PortEntityClient
 from .scorecards import PortScorecardClient
 
-__all__ = ["PortClient", "PortAgentClient", "PortBlueprintClient", "PortEntityClient", "PortScorecardClient"]
+__all__ = [
+    "PortClient",
+    "PortAgentClient",
+    "PortBlueprintClient",
+    "PortEntityClient",
+    "PortScorecardClient",
+    "PortActionClient",
+]

src/client/__init__.py

@@ -0,0 +1,33 @@
+from pyport import PortClient
+
+from src.models.action_run import ActionRun
+from src.utils import logger
+
+
+class PortActionRunClient:
+    def __init__(self, client: PortClient):
+        self._client = client
+
+    async def create_global_action_run(self, action_identifier: str, **kwargs) -> ActionRun:
+        logger.info(f"Creating global action run for: {action_identifier}")
+        response = self._client.make_request(
+            "POST", f"actions/{action_identifier}/runs", json=kwargs
+        )
+        action_run_data = response.json().get("run", response.json())
+        return ActionRun.construct(**action_run_data)
+
+    async def create_entity_action_run(self, action_identifier: str, **kwargs) -> ActionRun:
+        logger.info(f"Creating entity action run for {action_identifier} with kwargs: {kwargs}")
+        response = self._client.make_request(
+            "POST",
+            f"actions/{action_identifier}/runs",
+            json=kwargs,
+        )
+        action_run_data = response.json().get("run", response.json())
+        return ActionRun.construct(**action_run_data)
+
+    async def get_action_run(self, run_id: str) -> ActionRun:
+        logger.debug(f"Getting action run status for: {run_id}")
+        response = self._client.make_request("GET", f"actions/runs/{run_id}?version=v2")
+        action_run_data = response.json().get("run", response.json())
+        return ActionRun.construct(**action_run_data)

src/client/action_runs.py

@@ -0,0 +1,36 @@
+from pyport import PortClient
+
+from src.config import config
+from src.models.actions import Action
+from src.utils import logger
+
+
+class PortActionClient:
+    def __init__(self, client: PortClient):
+        self._client = client
+
+    async def get_all_actions(self, trigger_type: str = "self-service") -> list[Action]:
+        logger.info("Getting all actions")
+
+        response = self._client.make_request("GET", f"actions?trigger_type={trigger_type}")
+        result = response.json().get("actions", [])
+
+        if config.api_validation_enabled:
+            logger.debug("Validating actions")
+            return [Action(**action) for action in result]
+        else:
+            logger.debug("Skipping API validation for actions")
+            return [Action.construct(**action) for action in result]
+
+    async def get_action(self, action_identifier: str) -> Action:
+        logger.info(f"Getting action: {action_identifier}")
+
+        response = self._client.make_request("GET", f"actions/{action_identifier}")
+        result = response.json().get("action")
+
+        if config.api_validation_enabled:
+            logger.debug("Validating action")
+            return Action(**result)
+        else:
+            logger.debug("Skipping API validation for action")
+            return Action.construct(**result)

src/client/actions.py

@@ -67,6 +67,7 @@ async def get_invocation_status(self, identifier: str) -> PortAgentResponse:
                     error=None if status.lower() != "error" else message,
                     action_url=action_url,
                     selected_agent=selected_agent,
+                    raw_output=response_data,
                 )
             else:
                 return PortAgentResponse.construct(

src/client/agent.py

@@ -1,21 +1,27 @@
-from collections.abc import Callable
-from typing import Any
+from collections.abc import Awaitable, Callable
+from typing import Any, TypeVar
 
 import pyport
-import requests
+import requests  # type: ignore[import-untyped]
 
+from src.client.action_runs import PortActionRunClient
+from src.client.actions import PortActionClient
 from src.client.agent import PortAgentClient
 from src.client.blueprints import PortBlueprintClient
 from src.client.entities import PortEntityClient
 from src.client.scorecards import PortScorecardClient
 from src.config import config
+from src.models.action_run.action_run import ActionRun
+from src.models.actions.action import Action
 from src.models.agent import PortAgentResponse
 from src.models.agent.port_agent_response import PortAgentTriggerResponse
 from src.models.blueprints import Blueprint
 from src.models.entities import EntityResult
 from src.models.scorecards import Scorecard
 from src.utils import PortError, logger
 
+T = TypeVar("T")
+
 
 class PortClient:
     """Client for interacting with the Port API."""
@@ -27,36 +33,41 @@ def __init__(
         region: str = "EU",
         base_url: str = config.port_api_base,
     ):
+        if not client_id or not client_secret:
+            logger.warning("PortClient initialized without credentials")
+
         self.base_url = base_url
         self.client_id = client_id
         self.client_secret = client_secret
         self.region = region
-
-        if not client_id or not client_secret:
-            logger.warning("Port client initialized without credentials")
-            self._client = None
-            self.agent = None
-            self.blueprints = None
-            self.entities = None
-            self.scorecards = None
-        else:
-            self._client = pyport.PortClient(client_id=client_id, client_secret=client_secret, us_region=(region == "US"))
+        if client_id and client_secret:
+            self._client = pyport.PortClient(
+                client_id=client_id,
+                client_secret=client_secret,
+                us_region=(region == "US"),
+            )
             self.agent = PortAgentClient(self._client)
             self.blueprints = PortBlueprintClient(self._client)
             self.entities = PortEntityClient(self._client)
             self.scorecards = PortScorecardClient(self._client)
+            self.actions = PortActionClient(self._client)
+            self.action_runs = PortActionRunClient(self._client)
 
     def handle_http_error(self, e: requests.exceptions.HTTPError) -> PortError:
         result = e.response.json()
-        message = f"Error in {e.request.method} {e.request.url} - {e.response.status_code}: {result}"
+        message = (
+            f"Error in {e.request.method} {e.request.url} - {e.response.status_code}: {result}"
+        )
         logger.error(message)
         raise PortError(message)
 
-    async def wrap_request(self, request: Callable) -> PortError:
+    async def wrap_request(self, request: Callable[[], Awaitable[T]]) -> T:
+        if self._client is None:
+            raise PortError("PortClient is not properly initialized - missing credentials")
         try:
             return await request()
         except requests.exceptions.HTTPError as e:
-            self.handle_http_error(e)
+            raise self.handle_http_error(e) from e
 
     async def trigger_agent(self, prompt: str) -> PortAgentTriggerResponse:
         return await self.wrap_request(lambda: self.agent.trigger_agent(prompt))
@@ -77,36 +88,85 @@ async def update_blueprint(self, blueprint_data: dict[str, Any]) -> Blueprint:
         return await self.wrap_request(lambda: self.blueprints.update_blueprint(blueprint_data))
 
     async def delete_blueprint(self, blueprint_identifier: str) -> bool:
-        return await self.wrap_request(lambda: self.blueprints.delete_blueprint(blueprint_identifier))
+        return await self.wrap_request(
+            lambda: self.blueprints.delete_blueprint(blueprint_identifier)
+        )
 
     async def get_entity(self, blueprint_identifier: str, entity_identifier: str) -> EntityResult:
-        return await self.wrap_request(lambda: self.entities.get_entity(blueprint_identifier, entity_identifier))
+        return await self.wrap_request(
+            lambda: self.entities.get_entity(blueprint_identifier, entity_identifier)
+        )
 
     async def get_entities(self, blueprint_identifier: str) -> list[EntityResult]:
         return await self.wrap_request(lambda: self.entities.get_entities(blueprint_identifier))
 
-    async def create_entity(self, blueprint_identifier: str, entity_data: dict[str, Any], query: dict[str, Any]) -> EntityResult:
-        return await self.wrap_request(lambda: self.entities.create_entity(blueprint_identifier, entity_data, query))
+    async def create_entity(
+        self, blueprint_identifier: str, entity_data: dict[str, Any], query: dict[str, Any]
+    ) -> EntityResult:
+        return await self.wrap_request(
+            lambda: self.entities.create_entity(blueprint_identifier, entity_data, query)
+        )
 
-    async def update_entity(self, blueprint_identifier: str, entity_identifier: str, entity_data: dict[str, Any]) -> EntityResult:
-        return await self.wrap_request(lambda: self.entities.update_entity(blueprint_identifier, entity_identifier, entity_data))
+    async def update_entity(
+        self, blueprint_identifier: str, entity_identifier: str, entity_data: dict[str, Any]
+    ) -> EntityResult:
+        return await self.wrap_request(
+            lambda: self.entities.update_entity(
+                blueprint_identifier, entity_identifier, entity_data
+            )
+        )
 
-    async def delete_entity(self, blueprint_identifier: str, entity_identifier: str, delete_dependents: bool = False) -> bool:
+    async def delete_entity(
+        self, blueprint_identifier: str, entity_identifier: str, delete_dependents: bool = False
+    ) -> bool:
         return await self.wrap_request(
-            lambda: self.entities.delete_entity(blueprint_identifier, entity_identifier, delete_dependents)
+            lambda: self.entities.delete_entity(
+                blueprint_identifier, entity_identifier, delete_dependents
+            )
         )
 
     async def get_scorecard(self, blueprint_id: str, scorecard_id: str) -> Scorecard:
-        return await self.wrap_request(lambda: self.scorecards.get_scorecard(blueprint_id, scorecard_id))
+        return await self.wrap_request(
+            lambda: self.scorecards.get_scorecard(blueprint_id, scorecard_id)
+        )
 
     async def get_scorecards(self, blueprint_identifier: str) -> list[Scorecard]:
         return await self.wrap_request(lambda: self.scorecards.get_scorecards(blueprint_identifier))
 
-    async def create_scorecard(self, blueprint_id: str, scorecard_data: dict[str, Any]) -> Scorecard:
-        return await self.wrap_request(lambda: self.scorecards.create_scorecard(blueprint_id, scorecard_data))
+    async def create_scorecard(
+        self, blueprint_id: str, scorecard_data: dict[str, Any]
+    ) -> Scorecard:
+        return await self.wrap_request(
+            lambda: self.scorecards.create_scorecard(blueprint_id, scorecard_data)
+        )
 
-    async def update_scorecard(self, blueprint_id: str, scorecard_id: str, scorecard_data: dict[str, Any]) -> Scorecard:
-        return await self.wrap_request(lambda: self.scorecards.update_scorecard(blueprint_id, scorecard_id, scorecard_data))
+    async def update_scorecard(
+        self, blueprint_id: str, scorecard_id: str, scorecard_data: dict[str, Any]
+    ) -> Scorecard:
+        return await self.wrap_request(
+            lambda: self.scorecards.update_scorecard(blueprint_id, scorecard_id, scorecard_data)
+        )
 
     async def delete_scorecard(self, scorecard_id: str, blueprint_id: str) -> bool:
-        return await self.wrap_request(lambda: self.scorecards.delete_scorecard(scorecard_id, blueprint_id))
+        return await self.wrap_request(
+            lambda: self.scorecards.delete_scorecard(scorecard_id, blueprint_id)
+        )
+
+    async def get_all_actions(self, trigger_type: str = "self-service") -> list[Action]:
+        return await self.wrap_request(lambda: self.actions.get_all_actions(trigger_type))
+
+    async def get_action(self, action_identifier: str) -> Action:
+        return await self.wrap_request(lambda: self.actions.get_action(action_identifier))
+
+    async def create_global_action_run(self, action_identifier: str, **kwargs) -> ActionRun:
+        return await self.wrap_request(
+            lambda: self.action_runs.create_global_action_run(action_identifier, **kwargs)
+        )
+
+    async def create_entity_action_run(self, action_identifier: str, **kwargs) -> ActionRun:
+        return await self.wrap_request(
+            lambda: self.action_runs.create_entity_action_run(action_identifier, **kwargs)
+        )
+
+    async def get_action_run(self, run_id: str) -> ActionRun:
+        return await self.wrap_request(lambda: self.action_runs.get_action_run(run_id))

src/client/client.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, cast
 
 from pyport import PortClient
 
@@ -110,4 +110,4 @@ async def delete_entity(self, blueprint_identifier: str, entity_identifier: str,
             logger.warning(message)
             raise PortError(message)
         logger.info(f"Deleted entity '{entity_identifier}' from blueprint '{blueprint_identifier}' in Port")
-        return response_json.get("ok")
+        return cast(bool, response_json.get("ok"))