AI-2442: add read-only access control for guest and read-only roles

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "keboola-mcp-server"
-version = "1.44.7"
+version = "1.44.8"
 description = "MCP server for interacting with Keboola Connection"
 readme = "README.md"
 requires-python = ">=3.10"

src/keboola_mcp_server/authorization.py

@@ -22,7 +22,7 @@
 from fastmcp.tools import Tool
 from mcp import types as mt
 
-from keboola_mcp_server.mcp import get_http_request_or_none
+from keboola_mcp_server.mcp import get_http_request_or_none, is_read_only_tool
 
 LOG = logging.getLogger(__name__)
 
@@ -81,13 +81,6 @@ def _get_authorization_config() -> tuple[set[str] | None, set[str] | None, bool]
 
         return allowed_tools, disallowed_tools, read_only_mode
 
-    @staticmethod
-    def _is_read_only_tool(tool: Tool) -> bool:
-        """Check if a tool has readOnlyHint=True annotation."""
-        if tool.annotations is None:
-            return False
-        return tool.annotations.readOnlyHint is True
-
     @staticmethod
     def _is_tool_authorized(
         tool: Tool, allowed_tools: set[str] | None, disallowed_tools: set[str] | None, read_only_mode: bool
@@ -97,7 +90,7 @@ def _is_tool_authorized(
         if disallowed_tools and tool.name in disallowed_tools:
             return False
         # Check read-only mode - only allow tools with readOnlyHint=True
-        if read_only_mode and not ToolAuthorizationMiddleware._is_read_only_tool(tool):
+        if read_only_mode and not is_read_only_tool(tool):
             return False
         # Then check if tool is in allowed list (if specified)
         if allowed_tools is not None and tool.name not in allowed_tools:

src/keboola_mcp_server/cli.py

@@ -12,7 +12,6 @@
 from typing import Optional
 
 import pydantic
-import requests
 from fastmcp import FastMCP
 from starlette.exceptions import HTTPException
 from starlette.middleware import Middleware
@@ -95,7 +94,6 @@ async def _http_exception_handler(request: Request, exc: HTTPException):
 _exception_handlers = {
     HTTPException: _http_exception_handler,
     json.JSONDecodeError: _bad_request_handler,
-    requests.JSONDecodeError: _bad_request_handler,
     pydantic.ValidationError: _bad_request_handler,
     ValueError: _bad_request_handler,
     Exception: _create_exception_handler(status_code=500, log_exception=True),

src/keboola_mcp_server/errors.py

@@ -5,6 +5,7 @@
 from functools import wraps
 from typing import Any, Callable, Mapping, Optional, Type, TypeVar, cast
 
+import httpx
 import yaml
 from fastmcp import Context
 from fastmcp.exceptions import ToolError
@@ -170,7 +171,14 @@ async def wrapped(*args, **kwargs):
                     await _trigger_event(func, args, kwargs, exception, time.perf_counter() - start)
                 except Exception as e:
                     LOG.exception(f'Failed to trigger tool event for "{func.__name__}" tool: {e}')
-                    raise
+                    # Only swallow 403 Forbidden errors (expected for guest/read-only roles)
+                    # Re-raise other errors as they indicate genuine problems
+                    if isinstance(e, httpx.HTTPStatusError) and e.response.status_code == 403:
+                        # Expected for restricted roles (guest, read-only) - don't fail the tool call
+                        pass
+                    else:
+                        # Unexpected error - re-raise to alert about the problem
+                        raise
 
         return cast(F, wrapped)
 

src/keboola_mcp_server/mcp.py

@@ -44,6 +44,13 @@
 DEFAULT_CONCURRENCY = 10
 
 
+def is_read_only_tool(tool: Tool) -> bool:
+    """Check if a tool has readOnlyHint=True annotation."""
+    if tool.annotations is None:
+        return False
+    return tool.annotations.readOnlyHint is True
+
+
 @dataclasses.dataclass(frozen=True)
 class ServerState:
     config: Config
@@ -330,6 +337,10 @@ async def on_list_tools(
             # Filter out data app tools when the client is not using the main/production branch
             tools = [t for t in tools if t.name not in {'modify_data_app', 'get_data_apps', 'deploy_data_app'}]
 
+        if token_role == 'readonly':
+            tools = [t for t in tools if is_read_only_tool(t)]
+            LOG.debug(f'Read-only access: filtered to {len(tools)} read-only tools for role={token_role}')
+
         return tools
 
     async def on_call_tool(
@@ -342,6 +353,14 @@ async def on_call_tool(
         features = self.get_project_features(token_info)
         token_role = self.get_token_role(token_info).lower()
 
+        if token_role == 'readonly':
+            if not is_read_only_tool(tool):
+                raise ToolError(
+                    f'Access denied: The tool "{tool.name}" requires write permissions. '
+                    f'Your current role ({token_role}) only allows read-only operations. '
+                    f'Contact your administrator to request write access.'
+                )
+
         if 'hide-conditional-flows' in features:
             if tool.name == 'create_conditional_flow':
                 raise ToolError(
@@ -370,7 +389,7 @@ async def on_call_tool(
                     f'Use "{UPDATE_FLOW_TOOL_NAME}" to update flow configuration instead.'
                 )
 
-        if tool.name in {'modify_data_app', 'get_data_apps', 'deploy_data_app'}:
+        if tool.name in ('modify_data_app', 'get_data_apps', 'deploy_data_app'):
             if not self.is_client_using_main_branch(context.fastmcp_context):
                 raise ToolError('Data apps are supported only in the main production branch.')
 

tests/test_mcp.py

@@ -34,9 +34,13 @@ class NestedModel(BaseModel):
     field2: list[str] | None = None
 
 
-def _tool(name: str) -> MagicMock:
+def _tool(name: str, read_only: bool = False) -> MagicMock:
     tool = MagicMock()
     tool.name = name
+    if read_only:
+        tool.annotations.readOnlyHint = True
+    else:
+        tool.annotations = None
     return tool
 
 
@@ -410,27 +414,33 @@ async def call_next(_):
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
-        ('token_role', 'hidden_tool', 'visible_tool'),
+        ('token_role', 'hidden_tools', 'visible_tools'),
         [
-            ('admin', 'update_flow', 'modify_flow'),
-            ('share', 'update_flow', 'modify_flow'),
-            ('', 'modify_flow', 'update_flow'),
-            ('readOnly', 'modify_flow', 'update_flow'),
+            ('admin', {'update_flow'}, {'modify_flow', 'read_only_tool'}),
+            ('share', {'update_flow'}, {'modify_flow', 'read_only_tool'}),
+            ('', {'modify_flow'}, {'update_flow', 'read_only_tool'}),
+            ('readOnly', {'modify_flow', 'update_flow'}, {'read_only_tool'}),
+            ('guest', {'modify_flow'}, {'update_flow', 'read_only_tool'}),
         ],
     )
     async def test_list_tools_filters_flow_tools_by_role(
         self,
         mcp_context_client,
         token_role: str,
-        hidden_tool: str,
-        visible_tool: str,
+        hidden_tools: set[str],
+        visible_tools: set[str],
     ) -> None:
         keboola_client = KeboolaClient.from_state(mcp_context_client.session.state)
         keboola_client.storage_client.verify_token = AsyncMock(
             return_value={'owner': {'features': []}, 'admin': {'role': token_role}}
         )
 
-        tools = [_tool(hidden_tool), _tool(visible_tool), _tool('other_tool')]
+        tools = [
+            _tool('modify_flow'),
+            _tool('update_flow'),
+            _tool('other_tool'),
+            _tool('read_only_tool', read_only=True),
+        ]
 
         async def call_next(_):
             return tools
@@ -440,34 +450,41 @@ async def call_next(_):
         result = await middleware.on_list_tools(context, call_next)
 
         result_names = {t.name for t in result}
-        assert hidden_tool not in result_names
-        assert visible_tool in result_names
+        for tool_name in hidden_tools:
+            assert tool_name not in result_names
+        for tool_name in visible_tools:
+            assert tool_name in result_names
 
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
-        ('token_role', 'called_tool', 'expect_error'),
+        ('token_role', 'called_tool', 'tool_read_only', 'expect_error'),
         [
-            ('admin', 'modify_flow', False),
-            ('admin', 'update_flow', True),
-            ('share', 'modify_flow', False),
-            ('share', 'update_flow', True),
-            ('', 'modify_flow', True),
-            ('', 'update_flow', False),
+            ('admin', 'modify_flow', False, False),
+            ('admin', 'update_flow', False, True),
+            ('share', 'modify_flow', False, False),
+            ('share', 'update_flow', False, True),
+            ('', 'modify_flow', False, True),
+            ('', 'update_flow', False, False),
+            ('guest', 'write_tool', False, False),
+            ('guest', 'read_only_tool', True, False),
+            ('readOnly', 'write_tool', False, True),
+            ('readOnly', 'read_only_tool', True, False),
         ],
     )
     async def test_call_tool_blocks_flow_tools_by_role(
         self,
         mcp_context_client,
         token_role: str,
         called_tool: str,
+        tool_read_only: bool,
         expect_error: bool,
     ) -> None:
         keboola_client = KeboolaClient.from_state(mcp_context_client.session.state)
         keboola_client.storage_client.verify_token = AsyncMock(
             return_value={'owner': {'features': []}, 'admin': {'role': token_role}}
         )
 
-        tool = _tool(called_tool)
+        tool = _tool(called_tool, read_only=tool_read_only)
         mcp_context_client.fastmcp = SimpleNamespace(get_tool=AsyncMock(return_value=tool))
         context = SimpleNamespace(fastmcp_context=mcp_context_client, message=SimpleNamespace(name=called_tool))