fix(jira): filter Cloud-only tools on Server/DC (#1603)

Reported-by: AmirF194
Github-Issue: #1082

Author: FullYolk

src/mcp_atlassian/servers/jira.py

@@ -1899,7 +1899,7 @@ async def batch_create_issues(
 
 
 @jira_mcp.tool(
-    tags={"jira", "read", "toolset:jira_issues"},
+    tags={"jira", "read", "cloud_only", "toolset:jira_issues"},
     annotations={"title": "Batch Get Changelogs", "readOnlyHint": True},
 )
 async def batch_get_changelogs(
@@ -2385,7 +2385,7 @@ async def delete_issue(
 
 
 @jira_mcp.tool(
-    tags={"jira", "write", "toolset:jira_issues"},
+    tags={"jira", "write", "cloud_only", "toolset:jira_issues"},
     annotations={"title": "Move Issue to Project", "destructiveHint": True},
 )
 @check_write_access

src/mcp_atlassian/servers/main.py

@@ -259,18 +259,33 @@ def _tool_filter_context(self) -> dict[str, Any] | None:
         )
 
         header_based_services = {"jira": False, "confluence": False}
+        service_headers: dict[str, str] = {}
         request = getattr(req_context, "request", None)
         if request is not None:
-            service_headers = getattr(request.state, "atlassian_service_headers", {})
-            if service_headers:
+            request_service_headers = getattr(
+                request.state, "atlassian_service_headers", {}
+            )
+            if isinstance(request_service_headers, dict) and request_service_headers:
+                service_headers = request_service_headers
                 header_based_services = get_available_services(service_headers)
 
+        jira_config = (
+            app_lifespan_state.full_jira_config if app_lifespan_state else None
+        )
+        jira_url_header = service_headers.get("X-Atlassian-Jira-Url")
+        jira_is_cloud: bool | None = None
+        if jira_url_header:
+            jira_is_cloud = is_atlassian_cloud_url(jira_url_header)
+        elif jira_config is not None:
+            jira_is_cloud = bool(jira_config.is_cloud)
+
         return {
             "read_only": read_only,
             "enabled_tools_filter": enabled_tools_filter,
             "enabled_toolsets_filter": enabled_toolsets_filter,
             "app_lifespan_state": app_lifespan_state,
             "header_based_services": header_based_services,
+            "jira_is_cloud": jira_is_cloud,
         }
 
     def _is_tool_authorized(
@@ -294,13 +309,25 @@ def _is_tool_authorized(
             return False
         return True
 
+    @staticmethod
+    def _is_tool_supported_on_deployment(
+        tool_obj: FastMCPTool, ctx: dict[str, Any]
+    ) -> bool:
+        """Return whether a tool is supported by the configured deployment."""
+        tool_tags = tool_obj.tags
+        if "cloud_only" in tool_tags and "jira" in tool_tags:
+            return ctx["jira_is_cloud"] is not False
+        return True
+
     def _is_tool_enabled(
         self, registered_name: str, tool_obj: FastMCPTool, ctx: dict[str, Any]
     ) -> bool:
         """Listing filter: the tool is authorized AND its backing service is
         configured/available (the latter is a listing-only graceful-hide)."""
         if not self._is_tool_authorized(registered_name, tool_obj, ctx):
             return False
+        if not self._is_tool_supported_on_deployment(tool_obj, ctx):
+            return False
 
         app_lifespan_state = ctx["app_lifespan_state"]
         header_based_services = ctx["header_based_services"]
@@ -361,14 +388,18 @@ async def _list_tools_mcp(self) -> list[MCPTool]:
     async def _call_tool_mcp(self, key: str, arguments: dict[str, Any]) -> Any:
         # Enforce the same enablement filter at call time as at listing time, so a
         # tool hidden from the listing (read-only mode, not in ENABLED_TOOLS, toolset
-        # disabled, or service unavailable) cannot be invoked directly by name.
+        # disabled, or deployment-incompatible) cannot be invoked directly by name.
         # Under an active filter context, denials and genuinely unknown tools raise
         # byte-identical messages here (no exists-but-disabled leak), decoupled from
         # upstream's error format (FastMCP uses a repr-quoted name).
         ctx = self._tool_filter_context()
         if ctx is not None:
             tool_obj = await self.get_tool(key)
-            if tool_obj is None or not self._is_tool_authorized(key, tool_obj, ctx):
+            if (
+                tool_obj is None
+                or not self._is_tool_authorized(key, tool_obj, ctx)
+                or not self._is_tool_supported_on_deployment(tool_obj, ctx)
+            ):
                 raise NotFoundError(f"Unknown tool: {key}")
         return await super()._call_tool_mcp(key, arguments)
 

tests/unit/servers/test_mcp_protocol.py

@@ -26,6 +26,7 @@
     UserTokenMiddleware,
     health_check,
     main_lifespan,
+    main_mcp,
 )
 from tests.utils.factories import (
     ConfluencePageFactory,
@@ -35,6 +36,23 @@
 
 logger = logging.getLogger(__name__)
 
+JIRA_CLOUD_ONLY_TOOL_NAMES = {
+    "batch_get_changelogs",
+    "move_issue",
+}
+
+
+def _mock_tool(name, tags):
+    tool = MagicMock(spec=FastMCPTool)
+    tool.name = name
+    tool.tags = tags
+    tool.to_mcp_tool.return_value = MCPTool(
+        name=name,
+        description=f"Tool {name}",
+        inputSchema={"type": "object", "properties": {}},
+    )
+    return tool
+
 
 @pytest.mark.anyio
 class TestMCPProtocolIntegration:
@@ -100,15 +118,16 @@ async def test_call_tool_mcp_enforces_enablement_at_dispatch(
         """A tool filtered out of the listing must not be invocable by name.
 
         `_list_tools_mcp` hides read-only-excluded / non-enabled / disabled-toolset
-        tools, but the call path must re-check — otherwise a client can dispatch a
-        hidden tool directly by name. Verifies denied calls never reach the
-        underlying executor and enabled calls do (GHSA-3r68).
+        / deployment-incompatible tools, but the call path must re-check — otherwise
+        a client can dispatch a hidden tool directly by name. Verifies denied calls
+        never reach the underlying executor and enabled calls do (GHSA-3r68).
         """
         from unittest.mock import AsyncMock
 
         from fastmcp import FastMCP
         from fastmcp.exceptions import NotFoundError
 
+        mock_jira_config.is_cloud = False
         app_context = MainAppContext(
             full_jira_config=mock_jira_config,
             full_confluence_config=mock_confluence_config,
@@ -125,9 +144,12 @@ async def test_call_tool_mcp_enforces_enablement_at_dispatch(
         read_tool.tags = {"jira", "read"}
         write_tool = MagicMock(spec=FastMCPTool)
         write_tool.tags = {"jira", "write"}
+        cloud_only_tool = MagicMock(spec=FastMCPTool)
+        cloud_only_tool.tags = {"jira", "read", "cloud_only"}
         tools_by_name = {
             "jira_get_issue": read_tool,
             "jira_create_issue": write_tool,
+            "jira_batch_get_changelogs": cloud_only_tool,
         }
 
         async def mock_get_tool(name, version=None):
@@ -145,6 +167,13 @@ async def mock_get_tool(name, version=None):
                 await atlassian_mcp_server._call_tool_mcp("jira_create_issue", {})
             mock_super.assert_not_called()
 
+            # Cloud-only tool is deployment-excluded on Server/DC -> denied.
+            with pytest.raises(NotFoundError) as deployment_exc:
+                await atlassian_mcp_server._call_tool_mcp(
+                    "jira_batch_get_changelogs", {}
+                )
+            mock_super.assert_not_called()
+
             # Genuinely unknown tool -> denied by our override too (never reaches
             # super, whose repr-quoted message would leak tool existence).
             with pytest.raises(NotFoundError) as unknown_exc:
@@ -154,13 +183,64 @@ async def mock_get_tool(name, version=None):
             # Message parity: hidden-but-existing and genuinely unknown tools
             # produce the same unquoted format (no exists-but-disabled leak).
             assert str(denied_exc.value) == "Unknown tool: jira_create_issue"
+            assert (
+                str(deployment_exc.value) == "Unknown tool: jira_batch_get_changelogs"
+            )
             assert str(unknown_exc.value) == "Unknown tool: jira_no_such_tool"
 
             # Read tool is enabled -> passes the gate and reaches the executor.
             result = await atlassian_mcp_server._call_tool_mcp("jira_get_issue", {})
             assert result == "EXECUTED"
             mock_super.assert_called_once()
 
+    @pytest.mark.parametrize("is_cloud", [True, False], ids=["cloud", "server_dc"])
+    async def test_tool_filtering_by_jira_deployment(self, is_cloud):
+        """The production server advertises Cloud-only Jira tools only on Cloud."""
+        jira_config = MagicMock(spec=JiraConfig)
+        jira_config.is_cloud = is_cloud
+        app_context = MainAppContext(full_jira_config=jira_config)
+        request_context = MagicMock()
+        request_context.request = None
+        request_context.lifespan_context = {"app_lifespan_context": app_context}
+
+        with patch.object(main_mcp, "_mcp_server") as mcp_server:
+            mcp_server.request_context = request_context
+            listed_tool_names = {tool.name for tool in await main_mcp._list_tools_mcp()}
+
+        expected_cloud_only_tools = {
+            f"jira_{name}" for name in JIRA_CLOUD_ONLY_TOOL_NAMES
+        }
+        assert "jira_get_issue" in listed_tool_names
+        assert listed_tool_names & expected_cloud_only_tools == (
+            expected_cloud_only_tools if is_cloud else set()
+        )
+
+    async def test_tool_filtering_uses_header_based_jira_deployment(
+        self, atlassian_mcp_server
+    ):
+        """Per-request Jira URLs determine Cloud-only tool availability."""
+        app_context = MainAppContext()
+        request_context = MagicMock()
+        request_context.lifespan_context = {"app_lifespan_context": app_context}
+        request_context.request.state.atlassian_service_headers = {
+            "X-Atlassian-Jira-Personal-Token": "test-token",
+            "X-Atlassian-Jira-Url": "https://jira.example.com",
+        }
+        atlassian_mcp_server._mcp_server = MagicMock()
+        atlassian_mcp_server._mcp_server.request_context = request_context
+
+        tool = _mock_tool("jira_batch_get_changelogs", {"jira", "read", "cloud_only"})
+
+        async def mock_list_tools():
+            return [tool]
+
+        atlassian_mcp_server.list_tools = mock_list_tools
+
+        with MockEnvironment.clean_env():
+            tools = await atlassian_mcp_server._list_tools_mcp()
+
+        assert tools == []
+
     async def test_tool_discovery_with_full_configuration(
         self, atlassian_mcp_server, mock_jira_config, mock_confluence_config
     ):