Add action permissions and policies MCP tools for Port dynamic permissions (#45)

Author: Copilot

CHANGELOG.md

@@ -6,6 +6,15 @@ 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.18] - 2025-06-18
+
+### Added
+- Added `get_action_permissions` tool to fetch action RBAC and permissions configuration
+- Added `update_action_policies` tool to update action policies and dynamic permissions configuration
+- Added support for Port's dynamic permissions and policies through new MCP tools
+- Added `required_approval` field to `Action` model
+
 ## [0.2.17] - 2025-06-17
 
 ### Added

README.md

@@ -22,6 +22,12 @@ The [Port IO](https://www.getport.io/) MCP server is a [Model Context Protocol (
 - **Define rules** - "Add a rule that requires services to have a team owner to reach the Silver level"
 - **Setup quality gates** - "Create a rule that checks if services have proper documentation"
 
+### Manage Permissions & RBAC
+
+- **Fetch action permissions** - "What are the current permission settings for this action?"
+- **Update action policies** - "Configure approval workflows for the deployment action"
+- **Configure dynamic permissions** - "Set up team-based access control for this action"
+
 We're continuously expanding Port MCP's capabilities. Have a suggestion? We'd love to hear your feedback on our [roadmap](https://roadmap.getport.io/ideas)!
 
 # Installation

pyproject.toml

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

src/client/client.py

@@ -9,6 +9,7 @@
 from src.client.agent import PortAgentClient
 from src.client.blueprints import PortBlueprintClient
 from src.client.entities import PortEntityClient
+from src.client.permissions import PortPermissionsClient
 from src.client.scorecards import PortScorecardClient
 from src.config import config
 from src.models.action_run.action_run import ActionRun
@@ -22,8 +23,6 @@
 from src.utils.user_agent import get_user_agent
 
 T = TypeVar("T")
-
-
 class PortClient:
     """Client for interacting with the Port API."""
 
@@ -56,6 +55,7 @@ def __init__(
             self.scorecards = PortScorecardClient(self._client)
             self.actions = PortActionClient(self._client)
             self.action_runs = PortActionRunClient(self._client)
+            self.permissions = PortPermissionsClient(self._client)
 
     def _setup_custom_headers(self):
         """Setup custom headers for all HTTP requests."""
@@ -206,3 +206,9 @@ async def create_entity_action_run(self, action_identifier: str, **kwargs) -> Ac
 
     async def get_action_run(self, run_id: str) -> ActionRun:
         return await self.wrap_request(lambda: self.action_runs.get_action_run(run_id))
+
+    async def get_action_permissions(self, action_identifier: str) -> dict[str, Any]:
+        return await self.wrap_request(lambda: self.permissions.get_action_permissions(action_identifier))
+
+    async def update_action_policies(self, action_identifier: str, policies: dict[str, Any]) -> dict[str, Any]:
+        return await self.wrap_request(lambda: self.permissions.update_action_policies(action_identifier, policies))

src/client/permissions.py

@@ -0,0 +1,78 @@
+"""Client for Port permissions and RBAC operations."""
+
+from typing import Any
+
+from pyport import PortClient
+
+from src.utils import logger
+
+
+class PortPermissionsClient:
+    """Client for managing Port permissions and RBAC."""
+
+    def __init__(self, client: PortClient):
+        self._client = client
+
+    async def get_action_permissions(self, action_identifier: str) -> dict[str, Any]:
+        """Get permissions configuration for a specific action."""
+        logger.info(f"Getting permissions for action: {action_identifier}")
+        
+        try:
+            response = self._client.make_request("GET", f"actions/{action_identifier}/permissions")
+            result = response.json()
+            permissions = result.get("permissions", {})
+            logger.info(f"Permissions: {permissions}")
+            if result.get("ok"):
+                # Return the permissions data structure from the permissions endpoint
+                permissions_info = {
+                    "action_identifier": action_identifier,
+                    "executePermissions": permissions.get("execute", {}),
+                    "approvePermissions": permissions.get("approve", {}),
+                }
+                logger.debug(f"Retrieved action permissions: {permissions_info}")
+                return permissions_info
+            else:
+                logger.warning(f"Failed to get action permissions: {result}")
+                return {}
+        except Exception as e:
+            logger.error(f"Error getting action permissions: {e}")
+            return {}
+
+    async def update_action_policies(self, action_identifier: str, policies: dict[str, Any]) -> dict[str, Any]:
+        """Update policies configuration for a specific action."""
+        logger.info(f"Updating policies for action: {action_identifier}")
+        
+        try:
+            # Prepare the payload for updating policies - the policies should be sent directly
+            payload = policies
+            
+            response = self._client.make_request("PATCH", f"actions/{action_identifier}/permissions", json=payload)
+            result = response.json()
+            permissions = result.get("permissions", {})
+            if result.get("ok"):
+                updated_info = {
+                    "action_identifier": action_identifier,
+                    "updated_policies": {
+                        "execute": permissions.get("execute", {}),
+                        "approve": permissions.get("approve", {})
+                    },
+                    "success": True,
+                }
+                logger.info(f"Successfully updated policies for action: {action_identifier}")
+                return updated_info
+            else:
+                logger.warning(f"Failed to update action policies: {result}")
+                return {
+                    "action_identifier": action_identifier,
+                    "updated_policies": {},
+                    "success": False,
+                    "error": result.get("message", "Unknown error"),
+                }
+        except Exception as e:
+            logger.error(f"Error updating action policies: {e}")
+            return {
+                "action_identifier": action_identifier,
+                "updated_policies": {},
+                "success": False,
+                "error": str(e),
+            }

src/models/actions/action.py

@@ -135,6 +135,12 @@ class ActionCommon(BaseModel):
         alias="invocationMethod",
         serialization_alias="invocationMethod",
     )
+    required_approval: bool | SkipJsonSchema[None] = Field(
+        None,
+        description="Whether approval is required",
+        alias="requiredApproval",
+        serialization_alias="requiredApproval",
+    )
     approval_notification: dict[str, Any] | SkipJsonSchema[None] = Field(
         None,
         description="Approval notification configuration",

src/models/permissions/__init__.py

@@ -0,0 +1,11 @@
+"""Permission models for Port API interactions."""
+
+from .get_action_permissions import GetActionPermissionsToolResponse, GetActionPermissionsToolSchema
+from .update_action_policies import UpdateActionPoliciesToolResponse, UpdateActionPoliciesToolSchema
+
+__all__ = [
+    "GetActionPermissionsToolResponse",
+    "GetActionPermissionsToolSchema",
+    "UpdateActionPoliciesToolResponse",
+    "UpdateActionPoliciesToolSchema",
+]

src/models/permissions/get_action_permissions.py

@@ -0,0 +1,19 @@
+"""Get action permissions tool schemas."""
+
+from typing import Any
+
+from pydantic import Field
+
+from src.models.common.base_pydantic import BaseModel
+
+
+class GetActionPermissionsToolSchema(BaseModel):
+    """Schema for get action permissions tool."""
+    
+    action_identifier: str = Field(description="The identifier of the action to get permissions configuration for")
+
+
+class GetActionPermissionsToolResponse(BaseModel):
+    """Response model for get action permissions tool."""
+    
+    permissions: dict[str, Any] = Field(description="Permissions configuration for the action")

src/models/permissions/update_action_policies.py

@@ -0,0 +1,73 @@
+"""Update action policies tool schemas."""
+
+from typing import Any
+
+from pydantic import Field
+
+from src.models.common.base_pydantic import BaseModel
+
+
+class UpdateActionPoliciesToolSchema(BaseModel):
+    """Schema for update action policies tool."""
+    
+    action_identifier: str = Field(
+        description="The identifier of the action to update policies for"
+    )
+    policies: dict[str, Any] = Field(
+        description="""Policies configuration to update. This should contain the complete policies structure including:
+
+• **execute**: Execution permissions configuration
+  - roles: List of roles allowed to execute (e.g., ["Member", "Admin"])
+  - users: List of specific users allowed to execute
+  - teams: List of teams allowed to execute  
+  - ownedByTeam: Boolean indicating if team ownership is required
+  - policy: Dynamic policy with queries and conditions
+
+• **approve**: Approval workflow configuration
+  - roles: List of roles allowed to approve
+  - users: List of specific users allowed to approve
+  - teams: List of teams allowed to approve
+  - policy: Dynamic policy with queries and conditions for approval
+
+• **policy**: Dynamic conditions using queries and JQ expressions
+  - queries: Named queries to fetch entities/users
+  - conditions: JQ expressions that evaluate to true/false
+
+🔑 **CRITICAL IMPLEMENTATION NOTES:**
+
+1. **Team Queries**: Use "$team" meta property, not "team" regular property
+   - Correct: {"property": "$team", "operator": "containsAny", "value": ["team-name"]}
+   - Wrong: {"property": "team", "operator": "containsAny", "value": ["team-name"]}
+
+2. **Approval Conditions**: MUST return user identifier arrays, not booleans
+   - Correct: [.results.experts.entities[].identifier]
+   - Wrong: [.results.experts.entities[].identifier] | length > 0
+
+3. **User Team Membership Query Pattern**:
+   ```json
+   {
+     "rules": [
+       {"property": "$blueprint", "operator": "=", "value": "_user"},
+       {"property": "$team", "operator": "containsAny", "value": ["team-name"]}
+     ],
+     "combinator": "and"
+   }
+   ```
+
+Example structures based on Port's dynamic permissions:
+- Basic team-based: {"execute": {"roles": ["Member"], "teams": ["platform-team"]}}
+- Dynamic condition: {"execute": {"roles": ["Member"], "policy": {"queries": {...}, "conditions": [...]}}}
+- With approval workflow: {"execute": {...}, "approve": {"roles": ["Admin"], "policy": {...}}}
+- Prevent self-approval: Use policy conditions to exclude the executing user from approvers
+
+Supports Port's full dynamic permissions capabilities as described in the Port documentation."""
+    )
+
+
+
+class UpdateActionPoliciesToolResponse(BaseModel):
+    """Response model for update action policies tool."""
+    
+    action_identifier: str = Field(description="The action identifier that was updated")
+    updated_policies: dict[str, Any] = Field(description="The updated policies configuration")
+    success: bool = Field(description="Whether the update was successful")

src/tools/__init__.py

@@ -26,6 +26,10 @@
     GetEntityTool,
     UpdateEntityTool,
 )
+from src.tools.permissions import (
+    GetActionPermissionsTool,
+    UpdateActionPoliciesTool,
+)
 from src.tools.scorecard import (
     CreateScorecardTool,
     DeleteScorecardTool,
@@ -57,4 +61,6 @@
     "GetActionTool",
     "ListActionsTool",
     "TrackActionRunTool",
+    "GetActionPermissionsTool",
+    "UpdateActionPoliciesTool",
 ]