cypher - add read only config (#184)

* Add neo4j read only config to mcp cypher

* fix

* action store true

* docs

* update changelog, add unit and it for write tool availability, update config parsing logic for read only

* fix unit tests, update readme

* formatting

---------

Co-authored-by: alex <[email protected]>

Author: tomasonjo

servers/mcp-neo4j-cypher/CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Changed
 
 ### Added
+* Add env variable `NEO4J_READ_ONLY` and cli variable `--read-only` to configure tool availability
 
 ## v0.4.0
 

servers/mcp-neo4j-cypher/Makefile

@@ -1,3 +1,8 @@
+format:
+	uv run ruff check --select I . --fix
+	uv run ruff check --fix .
+	uv run ruff format .
+	
 install-dev:
 	uv sync
 

servers/mcp-neo4j-cypher/README.md

@@ -38,6 +38,7 @@ The server offers these core tools:
     - `query` (string): The Cypher update query
     - `params` (dictionary, optional): Parameters to pass to the Cypher query
   - Returns: A JSON serialized result summary counter with `{ nodes_updated: number, relationships_created: number, ... }`
+  - **Availability**: May be disabled by supplying --read-only as cli flag or `NEO4J_READ_ONLY=true` environment variable
 
 #### 🕸️ Schema Tools
 
@@ -405,6 +406,7 @@ docker run --rm -p 8000:8000 \
 | `NEO4J_MCP_SERVER_ALLOWED_HOSTS`   | `localhost,127.0.0.1`                   | Comma-separated list of allowed hosts (DNS rebinding protection) |
 | `NEO4J_RESPONSE_TOKEN_LIMIT`       | _(none)_                                | Maximum tokens for read query responses            |
 | `NEO4J_READ_TIMEOUT`               | `30`                                    | Timeout in seconds for read queries                |
+| `NEO4J_READ_ONLY`                  | `false`                                 | Allow only read-only queries (true/false)          |
 
 ### 🌐 SSE Transport for Legacy Web Access
 

servers/mcp-neo4j-cypher/manifest.json

@@ -160,6 +160,14 @@
       "default": 30,
       "required": false,
       "sensitive": false
+    },
+    "read_only": {
+      "type": "boolean",
+      "title": "Read Only Mode",
+      "description": "Allow only read-only queries, no write operations permitted.",
+      "default": false,
+      "required": false,
+      "sensitive": false
     }
   }
 }

servers/mcp-neo4j-cypher/src/mcp_neo4j_cypher/__init__.py

@@ -32,10 +32,15 @@ def main():
         help="Allowed hosts for DNS rebinding protection on remote servers(comma-separated list)",
     )
     parser.add_argument(
-        "--read-timeout", 
-        type=int, 
-        default=None, 
-        help="Timeout in seconds for read queries (default: 30)"
+        "--read-timeout",
+        type=int,
+        default=None,
+        help="Timeout in seconds for read queries (default: 30)",
+    )
+    parser.add_argument(
+        "--read-only",
+        action="store_true",
+        help="Allow only read-only queries (default: False)",
     )
     parser.add_argument("--token-limit", default=None, help="Response token limit")
 

servers/mcp-neo4j-cypher/src/mcp_neo4j_cypher/server.py

@@ -7,15 +7,14 @@
 from fastmcp.server import FastMCP
 from fastmcp.tools.tool import TextContent, ToolResult
 from mcp.types import ToolAnnotations
-from neo4j import AsyncDriver, AsyncGraphDatabase, RoutingControl, Query
+from neo4j import AsyncDriver, AsyncGraphDatabase, Query, RoutingControl
 from neo4j.exceptions import ClientError, Neo4jError
 from pydantic import Field
 from starlette.middleware import Middleware
 from starlette.middleware.cors import CORSMiddleware
 from starlette.middleware.trustedhost import TrustedHostMiddleware
 
-from .utils import _value_sanitize
-from .utils import _value_sanitize, _truncate_string_to_tokens
+from .utils import _truncate_string_to_tokens, _value_sanitize
 
 logger = logging.getLogger("mcp_neo4j_cypher")
 
@@ -39,17 +38,19 @@ def _is_write_query(query: str) -> bool:
 
 
 def create_mcp_server(
-    neo4j_driver: AsyncDriver, 
-    database: str = "neo4j", 
+    neo4j_driver: AsyncDriver,
+    database: str = "neo4j",
     namespace: str = "",
     read_timeout: int = 30,
     token_limit: Optional[int] = None,
+    read_only: bool = False,
 ) -> FastMCP:
     mcp: FastMCP = FastMCP(
         "mcp-neo4j-cypher", dependencies=["neo4j", "pydantic"], stateless_http=True
     )
 
     namespace_prefix = _format_namespace(namespace)
+    allow_writes = not read_only
 
     @mcp.tool(
         name=namespace_prefix + "get_neo4j_schema",
@@ -219,6 +220,7 @@ async def read_neo4j_cypher(
             idempotentHint=False,
             openWorldHint=True,
         ),
+        enabled=allow_writes,
     )
     async def write_neo4j_cypher(
         query: str = Field(..., description="The Cypher query to execute."),
@@ -272,6 +274,7 @@ async def main(
     allowed_hosts: list[str] = [],
     read_timeout: int = 30,
     token_limit: Optional[int] = None,
+    read_only: bool = False,
 ) -> None:
     logger.info("Starting MCP neo4j Server")
 
@@ -289,11 +292,12 @@ async def main(
             allow_methods=["GET", "POST"],
             allow_headers=["*"],
         ),
-        Middleware(TrustedHostMiddleware, 
-                   allowed_hosts=allowed_hosts)
+        Middleware(TrustedHostMiddleware, allowed_hosts=allowed_hosts),
     ]
-    
-    mcp = create_mcp_server(neo4j_driver, database, namespace, read_timeout, token_limit)
+
+    mcp = create_mcp_server(
+        neo4j_driver, database, namespace, read_timeout, token_limit, read_only
+    )
 
     # Run the server with the specified transport
     match transport:
@@ -311,7 +315,13 @@ async def main(
             logger.info(
                 f"Running Neo4j Cypher MCP Server with SSE transport on {host}:{port}..."
             )
-            await mcp.run_http_async(host=host, port=port, path=path, middleware=custom_middleware, transport="sse")
+            await mcp.run_http_async(
+                host=host,
+                port=port,
+                path=path,
+                middleware=custom_middleware,
+                transport="sse",
+            )
         case _:
             logger.error(
                 f"Invalid transport: {transport} | Must be either 'stdio', 'sse', or 'http'"

servers/mcp-neo4j-cypher/src/mcp_neo4j_cypher/utils.py

@@ -1,14 +1,47 @@
-import tiktoken
-
 import argparse
 import logging
 import os
 from typing import Any, Union
 
+import tiktoken
+
 logger = logging.getLogger("mcp_neo4j_cypher")
 logger.setLevel(logging.INFO)
 
 
+def parse_boolean_safely(value: Union[str, bool]) -> bool:
+    """
+    Safely parse a string value to boolean with strict validation.
+
+    Parameters
+    ----------
+    value : Union[str, bool]
+        The value to parse to boolean.
+
+    Returns
+    -------
+    bool
+        The parsed boolean value.
+    """
+
+    if isinstance(value, bool):
+        return value
+
+    elif isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized == "true":
+            return True
+        elif normalized == "false":
+            return False
+        else:
+            raise ValueError(
+                f"Invalid boolean value: '{value}'. Must be 'true' or 'false'"
+            )
+    # we shouldn't get here, but just in case
+    else:
+        raise ValueError(f"Invalid boolean value: '{value}'. Must be 'true' or 'false'")
+
+
 def process_config(args: argparse.Namespace) -> dict[str, Union[str, int, None]]:
     """
     Process the command line arguments and environment variables to create a config dictionary.
@@ -173,14 +206,17 @@ def process_config(args: argparse.Namespace) -> dict[str, Union[str, int, None]]
     # parse allow origins
     if args.allow_origins is not None:
         # Handle comma-separated string from CLI
-     
-        config["allow_origins"] = [origin.strip() for origin in args.allow_origins.split(",") if origin.strip()]
+
+        config["allow_origins"] = [
+            origin.strip() for origin in args.allow_origins.split(",") if origin.strip()
+        ]
 
     else:
         if os.getenv("NEO4J_MCP_SERVER_ALLOW_ORIGINS") is not None:
             # split comma-separated string into list
             config["allow_origins"] = [
-                origin.strip() for origin in os.getenv("NEO4J_MCP_SERVER_ALLOW_ORIGINS", "").split(",") 
+                origin.strip()
+                for origin in os.getenv("NEO4J_MCP_SERVER_ALLOW_ORIGINS", "").split(",")
                 if origin.strip()
             ]
         else:
@@ -192,21 +228,24 @@ def process_config(args: argparse.Namespace) -> dict[str, Union[str, int, None]]
     # parse allowed hosts for DNS rebinding protection
     if args.allowed_hosts is not None:
         # Handle comma-separated string from CLI
-        config["allowed_hosts"] = [host.strip() for host in args.allowed_hosts.split(",") if host.strip()]
-      
+        config["allowed_hosts"] = [
+            host.strip() for host in args.allowed_hosts.split(",") if host.strip()
+        ]
+
     else:
         if os.getenv("NEO4J_MCP_SERVER_ALLOWED_HOSTS") is not None:
             # split comma-separated string into list
             config["allowed_hosts"] = [
-                host.strip() for host in os.getenv("NEO4J_MCP_SERVER_ALLOWED_HOSTS", "").split(",") 
+                host.strip()
+                for host in os.getenv("NEO4J_MCP_SERVER_ALLOWED_HOSTS", "").split(",")
                 if host.strip()
             ]
         else:
             logger.info(
                 "Info: No allowed hosts provided. Defaulting to secure mode - only localhost and 127.0.0.1 allowed."
             )
             config["allowed_hosts"] = ["localhost", "127.0.0.1"]
-            
+
     # parse token limit
     if args.token_limit is not None:
         config["token_limit"] = args.token_limit
@@ -232,12 +271,31 @@ def process_config(args: argparse.Namespace) -> dict[str, Union[str, int, None]]
                 )
                 config["read_timeout"] = config["read_timeout"]
             except ValueError:
-                logger.warning("Warning: Invalid read timeout provided. Using default: 30 seconds")
+                logger.warning(
+                    "Warning: Invalid read timeout provided. Using default: 30 seconds"
+                )
                 config["read_timeout"] = 30
         else:
             logger.info("Info: No read timeout provided. Using default: 30 seconds")
             config["read_timeout"] = 30
 
+    # parse read-only
+    if args.read_only:
+        config["read_only"] = True
+        logger.info(
+            f"Info: Read-only mode set to {config['read_only']} via command line argument."
+        )
+    elif os.getenv("NEO4J_READ_ONLY") is not None:
+        config["read_only"] = parse_boolean_safely(os.getenv("NEO4J_READ_ONLY"))
+        logger.info(
+            f"Info: Read-only mode set to {config['read_only']} via environment variable."
+        )
+    else:
+        logger.info(
+            "Info: No read-only setting provided. Write queries will be allowed."
+        )
+        config["read_only"] = False
+
     return config
 
 
@@ -295,6 +353,7 @@ def _value_sanitize(d: Any, list_limit: int = 128) -> Any:
     else:
         return d
 
+
 def _truncate_string_to_tokens(
     text: str, token_limit: int, model: str = "gpt-4"
 ) -> str:

servers/mcp-neo4j-cypher/tests/integration/conftest.py

@@ -269,3 +269,51 @@ async def http_server_custom_hosts(setup: Neo4jContainer):
     except asyncio.TimeoutError:
         process.kill()
         await process.wait()
+
+
+@pytest_asyncio.fixture
+async def http_server_read_only(setup: Neo4jContainer):
+    """Start the MCP server in HTTP mode with read-only enabled."""
+    # Start server process in HTTP mode with read-only
+    process = await asyncio.create_subprocess_exec(
+        "uv",
+        "run",
+        "mcp-neo4j-cypher",
+        "--transport",
+        "http",
+        "--server-host",
+        "127.0.0.1",
+        "--server-port",
+        "8005",
+        "--read-only",
+        "--db-url",
+        setup.get_connection_url(),
+        "--username",
+        setup.username,
+        "--password",
+        setup.password,
+        env=os.environ.copy(),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        cwd=os.getcwd(),
+    )
+
+    # Wait for server to start
+    await asyncio.sleep(3)
+
+    # Check if process is still running
+    if process.returncode is not None:
+        stdout, stderr = await process.communicate()
+        raise RuntimeError(
+            f"Read-only server failed to start. stdout: {stdout.decode()}, stderr: {stderr.decode()}"
+        )
+
+    yield process
+
+    # Cleanup
+    try:
+        process.terminate()
+        await asyncio.wait_for(process.wait(), timeout=5.0)
+    except asyncio.TimeoutError:
+        process.kill()
+        await process.wait()