refactor: change allow-write to read-only

refactor: fix tests broken because of changing allowing-write to read-only

Author: Kyon Caldera

README.md

@@ -35,7 +35,7 @@ Add this to your MCP client configuration, replacing env variables to match the
 Optional arguments you can add:
 - `--service`: AWS service name for SigV4 signing (inferred from endpoint if not provided)
 - `--profile`: AWS profile to use (uses AWS_PROFILE environment variable if not provided)
-- `--allow-write`: Allow tools that require write permissions to be enabled (by default, only tools with the `readOnlyHint` annotation are enabled)
+- `--read-only`: Disable tools which require write permissions. (tools which DO NOT require write permissions are annotated `readOnlyHint`)
 
 NOTE: `remote-server-url` should be your remote mcp server's URL (including the `/mcp` part). `service-code` should be the service code for your own mcp service, such as `eks-mcp`.
 
@@ -57,7 +57,7 @@ Example with all options
         "<service-code>",
         "--profile",
         "default",
-        "--allow-write"
+        "--read-only"
       ]
     }
   }

aws_mcp_proxy/mcp_proxy_manager.py

@@ -23,15 +23,15 @@ class McpProxyManager:
 
     logger = logging.getLogger(__name__)
 
-    def __init__(self, target_mcp: FastMCP, allow_write: bool = False):
+    def __init__(self, target_mcp: FastMCP, read_only: bool = False):
         """Initialize the MCP Proxy Manager.
 
         Args:
             target_mcp: The target MCP server to add content to
-            allow_write: Whether to allow tools that require write permissions
+            read_only: Whether to allow tools that require write permissions
         """
         self.target_mcp = target_mcp
-        self.allow_write = allow_write
+        self.read_only = read_only
 
     async def add_proxy_content(self, proxy: FastMCP) -> None:
         """Add tools, resources, and prompts from proxy to MCP server.
@@ -68,7 +68,7 @@ async def _add_tools(self, proxy: FastMCP) -> None:
         for tool_name, tool in tools.items():
             # Check the tool annotations and disable if needed
             annotations = tool.annotations
-            if not self.allow_write:
+            if self.read_only:
                 # In readOnly mode, skip the tools with no readOnlyHint=True annotation
                 if annotations and not annotations.readOnlyHint or not annotations:
                     self.logger.info(f'Skipping tool {tool_name} needing write permissions')

aws_mcp_proxy/server.py

@@ -63,7 +63,7 @@ async def setup_mcp_mode(mcp: FastMCP, args) -> None:
     proxy = FastMCP.as_proxy(transport)
 
     # Use McpProxyManager to add proxy content
-    proxy_manager = McpProxyManager(mcp, args.allow_write)
+    proxy_manager = McpProxyManager(mcp, args.read_only)
     await proxy_manager.add_proxy_content(proxy)
 
 
@@ -81,7 +81,7 @@ def parse_args():
   aws-mcp-proxy https://eks-mcp.us-west-2.api.aws --service eks-mcp --profile default
 
   # Run with write permissions enabled
-  aws-mcp-proxy https://eks-mcp.us-west-2.api.aws --allow-write
+  aws-mcp-proxy https://eks-mcp.us-west-2.api.aws --read-only
         """,
     )
 
@@ -101,9 +101,9 @@ def parse_args():
     )
 
     parser.add_argument(
-        '--allow-write',
+        '--read-only',
         action='store_true',
-        help='Allow tools that require write permissions to be enabled',
+        help='Disable tools which require write permissions',
     )
 
     parser.add_argument(

tests/test_mcp_proxy_manager.py

@@ -69,19 +69,19 @@ def mock_prompt(self):
         prompt.copy.return_value = prompt
         return prompt
 
-    def test_init_default_allow_write(self, mock_target_mcp):
-        """Test McpProxyManager initialization with default allow_write."""
+    def test_init_default_read_only(self, mock_target_mcp):
+        """Test McpProxyManager initialization with default read_only."""
         manager = McpProxyManager(mock_target_mcp)
 
         assert manager.target_mcp == mock_target_mcp
-        assert manager.allow_write is False
+        assert manager.read_only is False
 
-    def test_init_custom_allow_write(self, mock_target_mcp):
-        """Test McpProxyManager initialization with custom allow_write."""
-        manager = McpProxyManager(mock_target_mcp, allow_write=True)
+    def test_init_custom_read_only(self, mock_target_mcp):
+        """Test McpProxyManager initialization with custom read_only."""
+        manager = McpProxyManager(mock_target_mcp, read_only=True)
 
         assert manager.target_mcp == mock_target_mcp
-        assert manager.allow_write is True
+        assert manager.read_only is True
 
     @pytest.mark.asyncio
     async def test_add_proxy_content_success(
@@ -93,7 +93,7 @@ async def test_add_proxy_content_success(
         mock_proxy.get_resources.return_value = {'test_resource': mock_resource}
         mock_proxy.get_prompts.return_value = {'test_prompt': mock_prompt}
 
-        manager = McpProxyManager(mock_target_mcp, allow_write=True)
+        manager = McpProxyManager(mock_target_mcp, read_only=True)
         await manager.add_proxy_content(mock_proxy)
 
         # Verify all methods were called
@@ -120,31 +120,31 @@ async def test_add_tools_success(self, mock_target_mcp, mock_proxy, mock_tool):
 
     @pytest.mark.asyncio
     async def test_add_tools_skip_write_tools(self, mock_target_mcp, mock_proxy, mock_tool):
-        """Test that tools requiring write permissions are skipped when allow_write=False."""
+        """Test that tools requiring write permissions are skipped when read_only=True."""
         # Setup tool with write permissions required
         annotations = MagicMock()
         annotations.readOnlyHint = False
         mock_tool.annotations = annotations
 
         mock_proxy.get_tools.return_value = {'write_tool': mock_tool}
 
-        manager = McpProxyManager(mock_target_mcp, allow_write=False)
+        manager = McpProxyManager(mock_target_mcp, read_only=True)
         await manager._add_tools(mock_proxy)
 
         # Verify tool was not added (skipped)
         mock_target_mcp.add_tool.assert_not_called()
 
     @pytest.mark.asyncio
-    async def test_add_tools_allow_write_tools(self, mock_target_mcp, mock_proxy, mock_tool):
-        """Test that tools requiring write permissions are added when allow_write=True."""
+    async def test_add_tools_not_read_only_tools(self, mock_target_mcp, mock_proxy, mock_tool):
+        """Test that tools requiring write permissions are added when read_only=False."""
         # Setup tool with write permissions required
         annotations = MagicMock()
         annotations.readOnlyHint = False
         mock_tool.annotations = annotations
 
         mock_proxy.get_tools.return_value = {'write_tool': mock_tool}
 
-        manager = McpProxyManager(mock_target_mcp, allow_write=True)
+        manager = McpProxyManager(mock_target_mcp, read_only=False)
         await manager._add_tools(mock_proxy)
 
         # Verify tool was added
@@ -160,7 +160,7 @@ async def test_add_tools_readonly_tools(self, mock_target_mcp, mock_proxy, mock_
 
         mock_proxy.get_tools.return_value = {'readonly_tool': mock_tool}
 
-        manager = McpProxyManager(mock_target_mcp, allow_write=False)
+        manager = McpProxyManager(mock_target_mcp, read_only=True)
         await manager._add_tools(mock_proxy)
 
         # Verify tool was added
@@ -276,27 +276,27 @@ async def test_add_proxy_content_prompts_exception_handled(
 
     @pytest.mark.asyncio
     async def test_add_tools_no_annotations(self, mock_target_mcp, mock_proxy, mock_tool):
-        """Test that tools with no annotations are skipped when allow_write=False."""
+        """Test that tools with no annotations are skipped when read_only=True."""
         # Setup tool with no annotations
         mock_tool.annotations = None
 
         mock_proxy.get_tools.return_value = {'no_annotations_tool': mock_tool}
 
-        manager = McpProxyManager(mock_target_mcp, allow_write=False)
+        manager = McpProxyManager(mock_target_mcp, read_only=True)
         await manager._add_tools(mock_proxy)
 
         # Verify tool was not added (skipped)
         mock_target_mcp.add_tool.assert_not_called()
 
     @pytest.mark.asyncio
     async def test_add_tools_empty_annotations(self, mock_target_mcp, mock_proxy, mock_tool):
-        """Test that tools with empty annotations are skipped when allow_write=False."""
+        """Test that tools with empty annotations are skipped when read_only=True."""
         # Setup tool with empty annotations
         mock_tool.annotations = {}
 
         mock_proxy.get_tools.return_value = {'empty_annotations_tool': mock_tool}
 
-        manager = McpProxyManager(mock_target_mcp, allow_write=False)
+        manager = McpProxyManager(mock_target_mcp, read_only=True)
         await manager._add_tools(mock_proxy)
 
         # Verify tool was not added (skipped)

tests/test_server.py

@@ -37,7 +37,7 @@ async def test_setup_mcp_mode(
         mock_args.endpoint = 'https://test.example.com'
         mock_args.service = 'test-service'
         mock_args.profile = None
-        mock_args.allow_write = False
+        mock_args.read_only = True
 
         # Mock the transport and proxy
         mock_transport = Mock()
@@ -56,7 +56,7 @@ async def test_setup_mcp_mode(
         # Assert
         mock_create_transport.assert_called_once()
         mock_as_proxy.assert_called_once_with(mock_transport)
-        mock_proxy_manager_class.assert_called_once_with(mock_mcp, False)
+        mock_proxy_manager_class.assert_called_once_with(mock_mcp, True)
         mock_proxy_manager.add_proxy_content.assert_called_once_with(mock_proxy)
 
     @patch('aws_mcp_proxy.server.McpProxyManager')
@@ -72,7 +72,7 @@ async def test_setup_mcp_mode_with_tools(
         mock_args.endpoint = 'https://test.example.com'
         mock_args.service = 'test-service'
         mock_args.profile = None
-        mock_args.allow_write = False
+        mock_args.read_only = True
 
         # Mock the transport and proxy
         mock_transport = Mock()
@@ -91,7 +91,7 @@ async def test_setup_mcp_mode_with_tools(
         # Assert
         mock_create_transport.assert_called_once()
         mock_as_proxy.assert_called_once_with(mock_transport)
-        mock_proxy_manager_class.assert_called_once_with(mock_mcp, False)
+        mock_proxy_manager_class.assert_called_once_with(mock_mcp, True)
         mock_proxy_manager.add_proxy_content.assert_called_once_with(mock_proxy)
 
     @patch('aws_mcp_proxy.server.McpProxyManager')
@@ -107,7 +107,7 @@ async def test_setup_mcp_mode_tool_registration_error(
         mock_args.endpoint = 'https://test.example.com'
         mock_args.service = 'test-service'
         mock_args.profile = None
-        mock_args.allow_write = False
+        mock_args.read_only = True
 
         # Mock the transport and proxy
         mock_transport = Mock()