chore: added cud for actions tools (#41)

* chore: added cud for actions tools

* bump: 0.2.15

* tests: added tests to cud on actions

* testS: fix tests

* chore: lint

Author: danielsinai

CHANGELOG.md

@@ -5,6 +5,13 @@ 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.15] - 2025-06-16
+
+### Added
+- Added `create_action` tool to create actions
+- Added `delete_action` tool to delete actions
+- Added `update_action` tool to update actions
+
 ## [0.2.14] - 2025-06-15
 
 ### Changed

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "mcp-server-port"
-version = "0.2.14"
+version = "0.2.15"
 authors = [
   { name = "Matan Grady", email = "[email protected]" }
 ]

src/client/actions.py

@@ -1,3 +1,6 @@
+import json
+from typing import Any
+
 from pyport import PortClient
 
 from src.config import config
@@ -68,3 +71,67 @@ async def get_action(self, action_identifier: str) -> Action:
         else:
             logger.debug("Skipping API validation for action")
             return Action.construct(**result)
+        
+    async def create_action(self, action_data: dict[str, Any]) -> Action:
+        """Create a new action"""
+        data_json = json.dumps(action_data)
+
+        logger.info("Creating action in Port")
+        logger.debug(f"Input from tool to create action: {data_json}")
+
+        response = self._client.make_request("POST", "actions", json=action_data)
+        result = response.json()
+        if not result.get("ok"):
+            message = f"Failed to create action: {result}"
+            logger.warning(message)
+        logger.info("Action created in Port")
+
+        result = result.get("action", {})
+
+        if config.api_validation_enabled:
+            logger.debug("Validating action")
+            action = Action(**result)
+        else:
+            logger.debug("Skipping API validation for action")
+            action = Action.construct(**result)
+        logger.debug(f"Response for create action: {action}")
+        return action
+
+    async def update_action(self, action_identifier: str, action_data: dict[str, Any]) -> Action:
+        """Update an existing action"""
+        data_json = json.dumps(action_data)
+
+        logger.info(f"Updating action '{action_identifier}' in Port")
+        logger.debug(f"Input from tool to update action: {data_json}")
+
+        response = self._client.make_request(
+            "PUT", f"actions/{action_identifier}", json=action_data
+        )
+        result = response.json()
+        if not result.get("ok"):
+            message = f"Failed to update action: {result}"
+            logger.warning(message)
+        logger.info(f"Action '{action_identifier}' updated in Port")
+
+        result = result.get("action", {})
+        if config.api_validation_enabled:
+            logger.debug("Validating action")
+            action = Action(**result)
+        else:
+            logger.debug("Skipping API validation for action")
+            action = Action.construct(**result)
+        logger.debug(f"Response for update action: {action}")
+        return action
+
+    async def delete_action(self, action_identifier: str) -> bool:
+        """Delete an action"""
+        logger.info(f"Deleting action '{action_identifier}' from Port")
+
+        response = self._client.make_request("DELETE", f"actions/{action_identifier}")
+        result = response.json()
+        if not result.get("ok"):
+            message = f"Failed to delete action: {result}"
+            logger.warning(message)
+        logger.info(f"Action '{action_identifier}' deleted from Port")
+
+        return True

src/client/client.py

@@ -157,7 +157,18 @@ async def get_all_actions(self, trigger_type: str = "self-service") -> list[Acti
 
     async def get_action(self, action_identifier: str) -> Action:
         return await self.wrap_request(lambda: self.actions.get_action(action_identifier))
+    
+    async def create_action(self, action_data: dict[str, Any]) -> Action:
+        return await self.wrap_request(lambda: self.actions.create_action(action_data))
 
+    async def update_action(self, action_identifier: str, action_data: dict[str, Any]) -> Action:
+        return await self.wrap_request(
+            lambda: self.actions.update_action(action_identifier, action_data)
+        )
+
+    async def delete_action(self, action_identifier: str) -> bool:
+        return await self.wrap_request(lambda: self.actions.delete_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)

src/models/actions/__init__.py

@@ -1,5 +1,5 @@
 """Action related data models for Port.io."""
 
-from .action import Action, ActionSummary
+from .action import Action, ActionCreate, ActionSummary, ActionUpdate
 
-__all__ = ["Action", "ActionSummary"]
+__all__ = ["Action", "ActionCreate", "ActionUpdate", "ActionSummary"]

src/models/actions/action.py

@@ -1,6 +1,6 @@
 """Port.io action model."""
 
-from typing import Any
+from typing import Any, Literal
 
 from pydantic import Field
 from pydantic.json_schema import SkipJsonSchema
@@ -25,76 +25,153 @@ class ActionTrigger(BaseModel):
 
     type: str = Field(..., description="The type of trigger")
     operation: str | SkipJsonSchema[None] = Field(
-        None, description="The operation type (CREATE, DAY_2, DELETE)"
+        None, description="The operation type (CREATE, DAY-2, DELETE)"
     )
     event: str | SkipJsonSchema[None] = Field(
         None, description="The event that triggers the action"
     )
     condition: dict[str, Any] | SkipJsonSchema[None] = Field(
         None, description="Conditions for the trigger"
     )
+    user_inputs: ActionSchema | SkipJsonSchema[None] = Field(
+        None, description="User input schema for the trigger", alias="userInputs"
+    )
+    blueprint_identifier: str | SkipJsonSchema[None] = Field(
+        None, description="The blueprint identifier for the trigger", alias="blueprintIdentifier"
+    )
+
+
+class ActionInvocationMethodGitHub(BaseModel):
+    """GitHub invocation method configuration."""
+
+    type: Literal["GITHUB"] = Field(..., description="The type of invocation method")
+    org: str = Field(..., description="GitHub organization")
+    repo: str = Field(..., description="GitHub repository")
+    workflow: str = Field(..., description="GitHub workflow filename")
+    omit_payload: bool | SkipJsonSchema[None] = Field(None, description="Whether to omit payload")
+    omit_user_inputs: bool | SkipJsonSchema[None] = Field(
+        None, description="Whether to omit user inputs"
+    )
+    report_workflow_status: bool | SkipJsonSchema[None] = Field(
+        None, description="Whether to report workflow status"
+    )
+
+
+class ActionInvocationMethodGitLab(BaseModel):
+    """GitLab invocation method configuration."""
+
+    type: Literal["GITLAB"] = Field(..., description="The type of invocation method")
+    project_name: str = Field(..., description="GitLab project name", alias="projectName")
+    group_name: str = Field(..., description="GitLab group name", alias="groupName")
+    agent: Literal[True] = Field(..., description="Agent must be true for GitLab")
+    omit_payload: bool | SkipJsonSchema[None] = Field(
+        None, description="Whether to omit payload", alias="omitPayload"
+    )
+    omit_user_inputs: bool | SkipJsonSchema[None] = Field(
+        None, description="Whether to omit user inputs", alias="omitUserInputs"
+    )
+    default_ref: str | SkipJsonSchema[None] = Field(
+        None, description="Default Git reference", alias="defaultRef"
+    )
+
+
+class ActionInvocationMethodAzureDevOps(BaseModel):
+    """Azure DevOps invocation method configuration."""
 
+    type: Literal["AZURE-DEVOPS"] = Field(..., description="The type of invocation method")
+    org: str = Field(..., description="Azure DevOps organization")
+    webhook: str = Field(..., description="Azure DevOps webhook URL")
 
-class ActionInvocationMethod(BaseModel):
-    """Action invocation method configuration."""
 
-    type: str = Field(..., description="The type of invocation method")
-    url: str | SkipJsonSchema[None] = Field(None, description="URL for webhook invocation")
+class ActionInvocationMethodWebhook(BaseModel):
+    """Webhook invocation method configuration."""
+
+    type: Literal["WEBHOOK"] = Field(..., description="The type of invocation method")
+    url: str = Field(..., description="Webhook URL")
     agent: bool | SkipJsonSchema[None] = Field(
         None, description="Whether to use agent for invocation"
     )
-    method: str | SkipJsonSchema[None] = Field(None, description="HTTP method for webhook")
+    synchronized: bool | SkipJsonSchema[None] = Field(
+        None, description="Whether the webhook is synchronized"
+    )
+    method: Literal["POST", "DELETE", "PATCH", "PUT"] | SkipJsonSchema[None] = Field(
+        None, description="HTTP method for webhook"
+    )
     headers: dict[str, str] | SkipJsonSchema[None] = Field(None, description="Headers for webhook")
     body: str | dict[str, Any] | SkipJsonSchema[None] = Field(
         None, description="Body template for webhook (can be string or dict)"
     )
 
 
-class ActionSummary(BaseModel):
-    """Simplified Action model with only basic information."""
+class ActionInvocationMethodKafka(BaseModel):
+    """Kafka invocation method configuration."""
 
-    identifier: str = Field(..., description="The unique identifier of the action")
-    title: str = Field(..., description="The title of the action")
-    description: str | SkipJsonSchema[None] = Field(
-        None, description="The description of the action"
-    )
-    blueprint: str | SkipJsonSchema[None] = Field(
-        None, description="The blueprint this action belongs to"
-    )
+    type: Literal["KAFKA"] = Field(..., description="The type of invocation method")
 
 
-class Action(BaseModel):
-    """Port.io Action model."""
+# Union type for all invocation methods
+ActionInvocationMethod = (
+    ActionInvocationMethodGitHub
+    | ActionInvocationMethodGitLab
+    | ActionInvocationMethodAzureDevOps
+    | ActionInvocationMethodWebhook
+    | ActionInvocationMethodKafka
+)
+
+
+class ActionCommon(BaseModel):
+    """Common fields for action models."""
 
     identifier: str = Field(..., description="The unique identifier of the action")
     title: str = Field(..., description="The title of the action")
     description: str | SkipJsonSchema[None] = Field(
         None, description="The description of the action"
     )
     icon: Icon | SkipJsonSchema[None] = Field(None, description="The icon of the action")
-    blueprint: str | SkipJsonSchema[None] = Field(
-        None, description="The blueprint this action belongs to"
-    )
     trigger: ActionTrigger = Field(..., description="The trigger configuration")
     invocation_method: ActionInvocationMethod = Field(
         ...,
         description="The invocation method configuration",
         alias="invocationMethod",
         serialization_alias="invocationMethod",
     )
-    user_inputs: ActionSchema = Field(
-        default_factory=ActionSchema,
-        description="User input schema for the action",
-        alias="userInputs",
-        serialization_alias="userInputs",
-    )
     approval_notification: dict[str, Any] | SkipJsonSchema[None] = Field(
         None,
         description="Approval notification configuration",
         alias="approvalNotification",
         serialization_alias="approvalNotification",
     )
+
+
+class ActionSummary(BaseModel):
+    """Simplified Action model with only basic information."""
+
+    identifier: str = Field(..., description="The unique identifier of the action")
+    title: str = Field(..., description="The title of the action")
+    description: str | SkipJsonSchema[None] = Field(
+        None, description="The description of the action"
+    )
+    blueprint: str | SkipJsonSchema[None] = Field(
+        None, description="The blueprint this action belongs to"
+    )
+
+
+class Action(ActionCommon):
+    """Port.io Action model."""
+
     created_at: str | SkipJsonSchema[None] = Field(None, description="Creation timestamp")
     created_by: str | SkipJsonSchema[None] = Field(None, description="Creator user")
     updated_at: str | SkipJsonSchema[None] = Field(None, description="Last update timestamp")
     updated_by: str | SkipJsonSchema[None] = Field(None, description="Last updater user")
+
+
+class ActionCreate(ActionCommon):
+    """Model for creating a new action."""
+
+    pass
+
+
+class ActionUpdate(ActionCommon):
+    """Model for updating an existing action."""
+
+    pass

src/tools/__init__.py

@@ -4,9 +4,12 @@
 """
 
 from src.tools.action import (
+    CreateActionTool,
+    DeleteActionTool,
     GetActionTool,
     ListActionsTool,
     TrackActionRunTool,
+    UpdateActionTool,
 )
 from src.tools.ai_agent import InvokeAIAGentTool
 from src.tools.blueprint import (
@@ -32,6 +35,9 @@
 )
 
 __all__ = [
+    "CreateActionTool",
+    "DeleteActionTool",
+    "UpdateActionTool",
     "CreateScorecardTool",
     "DeleteScorecardTool",
     "GetScorecardTool",

src/tools/action/__init__.py

@@ -1,9 +1,15 @@
+from .create_action import CreateActionTool
+from .delete_action import DeleteActionTool
 from .dynamic_actions import DynamicActionToolsManager
 from .get_action import GetActionTool
 from .list_actions import ListActionsTool
 from .track_action_run import TrackActionRunTool
+from .update_action import UpdateActionTool
 
 __all__ = [
+    "CreateActionTool",
+    "DeleteActionTool",
+    "UpdateActionTool",
     "GetActionTool",
     "ListActionsTool",
     "TrackActionRunTool",

src/tools/action/create_action.py

@@ -0,0 +1,42 @@
+from typing import Any
+
+from src.client.client import PortClient
+from src.models.actions import Action, ActionCreate
+from src.models.common.annotations import Annotations
+from src.models.tools.tool import Tool
+
+
+class CreateActionToolSchema(ActionCreate):
+    pass
+
+
+class CreateActionTool(Tool[CreateActionToolSchema]):
+    port_client: PortClient
+
+    def __init__(self, port_client: PortClient):
+        super().__init__(
+            name="create_action",
+            description="Create a new self-service action or automation in your Port account. To learn more about actions and automations, check out the documentation at https://docs.port.io/actions-and-automations/",
+            function=self.create_action,
+            input_schema=CreateActionToolSchema,
+            output_schema=Action,
+            annotations=Annotations(
+                title="Create Action",
+                readOnlyHint=False,
+                destructiveHint=False,
+                idempotentHint=False,
+                openWorldHint=True,
+            ),
+        )
+        self.port_client = port_client
+
+    async def create_action(self, props: CreateActionToolSchema) -> dict[str, Any]:
+        """
+        Create a new action or automation.
+        """
+        action_data = props.model_dump(exclude_none=True, exclude_unset=True)
+
+        created_action = await self.port_client.create_action(action_data)
+        created_action_dict = created_action.model_dump(exclude_unset=True, exclude_none=True)
+
+        return created_action_dict