[PFR-4635] Tell MCP clients which tools only read

src/components/customizers.py

@@ -6,13 +6,16 @@
     OpenAPITool,
 )
 from fastmcp.utilities.openapi.models import HTTPRoute
+from mcp.types import ToolAnnotations
 
 from ..utils.logging import get_logger
 
 logger = get_logger(__name__)
 
 ComponentType = OpenAPITool | OpenAPIResource | OpenAPIResourceTemplate
 
+ACCESS_VALUES = ("read", "additive", "destructive")
+
 def customize_components(
     route: HTTPRoute,
     component: ComponentType,
@@ -51,6 +54,11 @@ def customize_components(
         # Hide this fully for now.
         component.output_schema = None
 
+        # Without a title, FastMCP derives one from the operationId: "Addcustomdataforentity".
+        if route.summary:
+            component.title = route.summary
+        component.annotations = _tool_annotations(route)
+
         # SCHEMA REFERENCE HANDLING:
         # OpenAPI-generated MCP tool schemas contain $defs with nested $ref references
         # (e.g., $ref -> $defs -> $ref chains).
@@ -100,6 +108,32 @@ def customize_components(
     logger.debug(f"Final tags: {component.tags}")
 
 
+def _tool_annotations(route: HTTPRoute) -> ToolAnnotations:
+    """Hint whether a tool only reads, only adds, or may overwrite data.
+
+    The x-cortex-mcp-access extension ("read", "additive" or "destructive") wins,
+    and any other value counts as destructive. Without it a GET reads and anything
+    else may be destructive, which is also what MCP clients assume.
+    """
+    method = route.method.upper()
+    access = route.extensions.get("x-cortex-mcp-access")
+    if access is None:
+        access = "read" if method == "GET" else "destructive"
+    elif access not in ACCESS_VALUES:
+        logger.warning(
+            f"Unknown x-cortex-mcp-access {access!r} on {method} {route.path}, "
+            "treating it as destructive"
+        )
+        access = "destructive"
+    read_only = access == "read"
+    # MCP defines destructiveHint only for tools that aren't read-only.
+    destructive = None if read_only else access == "destructive"
+    return ToolAnnotations(
+        read_only_hint=read_only,
+        destructive_hint=destructive,
+    )
+
+
 def _format_tool_description(route: HTTPRoute, original: str) -> str:
     """Format description for tool components."""
     method = route.method.upper()

swagger.json

@@ -9611,6 +9611,7 @@
         "tags": [
           "Eng Intel: 2-Metrics"
         ],
+        "x-cortex-mcp-access": "read",
         "x-cortex-mcp-description": "Execute point-in-time queries for one or more engineering metrics.\n\n        Returns current metric values for specified time periods, with support for batch queries\n        and optional period-over-period comparisons. Time range (startTime/endTime) cannot exceed 6 months (180 days).\n\n        PREREQUISITES - Follow this workflow:\n        1. Discover all available metrics ONCE: Call listMetricDefinitions (view\u003d\u0027basic\u0027) - cache this response\n        2. Get metric query metadata ONCE per metric: Call listMetricDefinitions (view\u003d\u0027full\u0027, key\u003dMETRIC_KEY)\n           - supportedAggregations: Valid aggregation methods\n           - orderByAttribute: Attribute path for sorting by metric values\n           - groupByOptions[].key: Valid groupBy keys (use exact values, do NOT guess)\n           - filterOptions[].key: Valid filter keys (use exact values, do NOT guess)\n           Cache the full view response for each metric. Reuse the metadata from cached responses for subsequent queries on the same metric.\n        3. Construct query: Use the query metadata from the full view responses in step 2 to build valid point-in-time requests\n\n        IMPORTANT: Cache only results from listMetricDefinitions. Do NOT cache point-in-time query results - always execute fresh queries for current data.\n        Only refresh cached listMetricDefinitions responses if no longer in your context window or explicitly requested.\n        Do NOT guess attribute names - always use exact values from listMetricDefinitions responses.\n\n        Response includes:\n        - Lightweight metadata: Column definitions optimized for programmatic use\n        - Row data: Actual metric values and dimensional data\n        - No heavy schemas: Source definitions excluded (get from listMetricDefinitions instead)\n\n        Error responses:\n        - 400: Invalid metric names, date range, validation errors, or unsupported metric combinations\n        - 403: Feature not enabled (contact help@cortex.io)",
         "x-cortex-mcp-enabled": "true",
         "x-internal": true
@@ -12410,6 +12411,7 @@
         "tags": [
           "My Workspace"
         ],
+        "x-cortex-mcp-access": "read",
         "x-cortex-mcp-description": "TOOL for retrieving current user\u0027s owned resources and work items across the Cortex workspace.\n\n        FLEXIBLE REQUEST STRUCTURE:\n        The request accepts an object with optional fields for each resource type:\n        - myEntitiesRequest: Fetch entities (services, resources, domains) owned by the user\n        - myTeamsRequest: Fetch teams the user belongs to\n        - myScorecardsRequest: Fetch scorecards associated with user\u0027s entities\n        - myOpenPRsRequest: Fetch user\u0027s open pull requests across all Git repositories\n        - myRequestedReviewsRequest: Fetch PRs where user is requested as a reviewer\n        - myWorkItemsRequest: Fetch work items (Jira, Linear, Azure DevOps) assigned to the user\n\n        Set one or more fields to fetch multiple resource types in a single call.\n        All results are combined into a single response array with mixed item types.\n\n        RESPONSE STRUCTURE:\n        Returns a single array of items where each item has an \u0027itemType\u0027 field indicating its type:\n        - \u0027entity\u0027: Catalog entities (services, resources, domains)\n        - \u0027team\u0027: Teams the user belongs to\n        - \u0027scorecard\u0027: Scorecards for user\u0027s entities\n        - \u0027pull_request\u0027: Open pull requests authored by the user\n        - \u0027review_request\u0027: Pull requests where user is requested as reviewer\n        - \u0027work_item\u0027: Work items (issues/tickets) assigned to the user\n\n        EXAMPLES:\n        1. Fetch only services:\n           {\"myEntitiesRequest\": {\"entityTypes\": [\"service\"]}}\n\n        2. Fetch teams and their entities:\n           {\"myTeamsRequest\": {\"includeMembers\": true}, \"myEntitiesRequest\": {}}\n\n        3. Get all work (PRs, reviews, Jira tickets):\n           {\"myOpenPRsRequest\": {}, \"myRequestedReviewsRequest\": {}, \"myWorkItemsRequest\": {}}\n\n        4. Search entities with filters:\n           {\"myEntitiesRequest\": {\"query\": \"payment\", \"ownershipSource\": \"DIRECT\", \"entityTypes\": [\"service\", \"resource\"]}}\n        ",
         "x-cortex-mcp-enabled": "true",
         "x-internal": true

tests/conftest.py

@@ -27,6 +27,7 @@ def sample_route():
     route = Mock()
     route.path = "/api/v1/resources"
     route.method = "GET"
+    route.summary = "List resources"
     route.tags = ["api", "resources"]
     route.extensions = {"x-cortex-mcp-enabled": "true"}
     return route

tests/test_components.py

@@ -1,6 +1,8 @@
 """Tests for component customization."""
+import logging
 from unittest.mock import Mock
 
+import pytest
 from fastmcp.server.providers.openapi import (
     OpenAPIResource,
     OpenAPIResourceTemplate,
@@ -87,3 +89,71 @@ def test_preserve_original_description(self, sample_route):
 
         assert original in tool.description
         assert tool.description != original
+
+
+class TestToolAnnotations:
+    """Hints that let clients approve read tools apart from write tools."""
+
+    @staticmethod
+    def _tool(route):
+        tool = Mock(spec=OpenAPITool)
+        tool.description = ""
+        tool.tags = set()
+        customize_components(route, tool)
+        return tool
+
+    def test_get_is_read_only(self, sample_route):
+        """Test that a GET without x-cortex-mcp-access is read-only."""
+        assert self._tool(sample_route).annotations.read_only_hint is True
+
+    @pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
+    def test_other_methods_default_to_destructive(self, sample_route, method):
+        """Test that other methods without x-cortex-mcp-access are destructive."""
+        sample_route.method = method
+
+        annotations = self._tool(sample_route).annotations
+
+        assert annotations.read_only_hint is False
+        assert annotations.destructive_hint is True
+        assert annotations.idempotent_hint is None
+
+    def test_access_extension_marks_a_post_read_only(self, sample_route):
+        """Test that x-cortex-mcp-access: read makes a POST read-only."""
+        sample_route.method = "POST"
+        sample_route.extensions = {"x-cortex-mcp-access": "read"}
+
+        assert self._tool(sample_route).annotations.read_only_hint is True
+
+    def test_access_extension_marks_a_write_additive(self, sample_route):
+        """Test that x-cortex-mcp-access: additive is a non-destructive write."""
+        sample_route.method = "POST"
+        sample_route.extensions = {"x-cortex-mcp-access": "additive"}
+
+        annotations = self._tool(sample_route).annotations
+
+        assert annotations.read_only_hint is False
+        assert annotations.destructive_hint is False
+
+    def test_access_extension_marks_a_get_destructive(self, sample_route):
+        """Test that x-cortex-mcp-access: destructive overrides the GET default."""
+        sample_route.extensions = {"x-cortex-mcp-access": "destructive"}
+
+        annotations = self._tool(sample_route).annotations
+
+        assert annotations.read_only_hint is False
+        assert annotations.destructive_hint is True
+
+    def test_unknown_access_is_destructive(self, sample_route, caplog):
+        """Test that an unknown x-cortex-mcp-access value warns and is destructive."""
+        caplog.set_level(logging.WARNING, logger="src.components.customizers")
+        sample_route.extensions = {"x-cortex-mcp-access": "Read"}
+
+        annotations = self._tool(sample_route).annotations
+
+        assert annotations.read_only_hint is False
+        assert annotations.destructive_hint is True
+        assert "Unknown x-cortex-mcp-access 'Read'" in caplog.text
+
+    def test_title_is_the_summary(self, sample_route):
+        """Test that the tool title is the operation summary."""
+        assert self._tool(sample_route).title == "List resources"

tests/test_spec.py

@@ -0,0 +1,34 @@
+"""Tests for the bundled OpenAPI spec."""
+import json
+from pathlib import Path
+
+from src.components.customizers import ACCESS_VALUES
+
+SPEC_PATH = Path(__file__).parent.parent / "swagger.json"
+
+
+class TestBundledSpec:
+    """Test suite for the swagger.json the server ships with."""
+
+    def test_mcp_tools_declare_valid_access(self):
+        """Test that MCP-enabled operations set a valid x-cortex-mcp-access.
+
+        Every operation other than a GET must set it. Without it, or with an
+        unknown value, the tool is published as destructive even if it only reads.
+        """
+        spec = json.loads(SPEC_PATH.read_text())
+
+        invalid = [
+            f"{method.upper()} {path}"
+            for path, item in spec["paths"].items()
+            for method, operation in item.items()
+            if isinstance(operation, dict)
+            and operation.get("x-cortex-mcp-enabled") == "true"
+            and (method != "get" or "x-cortex-mcp-access" in operation)
+            and operation.get("x-cortex-mcp-access") not in ACCESS_VALUES
+        ]
+
+        assert not invalid, (
+            f"Set x-cortex-mcp-access to one of {', '.join(ACCESS_VALUES)} on: "
+            f"{', '.join(invalid)}"
+        )