feat: --read-only flag to disable write tools

README.md

@@ -186,6 +186,7 @@ The `mcp-bugzilla` command supports the following options:
 | `--port <PORT>` | `MCP_PORT` | `8000` | Port for the MCP server to listen on |
 | `--api-key-header <HEADER_NAME>` | `MCP_API_KEY_HEADER` | `ApiKey` | HTTP header name for the Bugzilla API key |
 | `--use-auth-header` | `USE_AUTH_HEADER` | `False` | Use `Authorization: Bearer` header instead of `api_key` query parameter |
+| `--read-only` | N/A | False | Disables all tools which can modify a bug. Works well in conjunction with `MCP_BUGZILLA_DISABLED_METHODS`
 
 **Note**: Command-line arguments take precedence over environment variables.
 
@@ -199,8 +200,8 @@ export MCP_HOST=127.0.0.1
 export MCP_PORT=8000
 export MCP_API_KEY_HEADER=ApiKey
 export LOG_LEVEL=INFO  # Optional: DEBUG, INFO, WARNING, ERROR, CRITICAL
-
 # Selective Disabling Tools (Optional)
+# Works fine in conjunction with --read-only flag
 export MCP_BUGZILLA_DISABLED_METHODS='bug_info,bug_comments'
 
 mcp-bugzilla

src/mcp_bugzilla/__init__.py

@@ -4,6 +4,8 @@
 import os
 import sys
 
+from pydantic_core import ArgsKwargs
+
 from . import server
 from .mcp_utils import mcp_log
 
@@ -40,9 +42,14 @@ def main():
     parser.add_argument(
         "--use-auth-header",
         action="store_true",
-        help="Use Authorization: Bearer header instead of api_key query parameter (required for some Bugzilla instances)"
+        help="Use Authorization: Bearer header instead of api_key query parameter (required for some Bugzilla instances)",
     )
 
+    parser.add_argument(
+        "--read-only",
+        action="store_true",
+        help="Disables all methods which modify the state of the bug",
+    )
     args = parser.parse_args()
 
     # The default behavior of argparse with os.getenv already handles the priority:
@@ -54,12 +61,7 @@ def main():
         )
         sys.exit(1)
 
-    server.cli_args["bugzilla_server"] = args.bugzilla_server
-    server.cli_args["host"] = args.host
-    server.cli_args["port"] = args.port
-    server.cli_args["api_key_header"] = args.api_key_header
-    server.cli_args["use_auth_header"] = args.use_auth_header
-
+    server.cli_args = args
     server.start()
 
 

src/mcp_bugzilla/server.py

@@ -7,24 +7,22 @@
 """
 
 import importlib.metadata
-from typing import Any, List
-from datetime import datetime
-
+from argparse import Namespace
 from contextlib import asynccontextmanager
+from datetime import datetime
+from typing import Any, List, Optional
 
-import httpx
 from fastmcp import FastMCP
 from fastmcp.dependencies import CurrentHeaders, Depends
 from fastmcp.exceptions import PromptError, ResourceError, ToolError, ValidationError
-from typing import Optional
 
 from .mcp_utils import Bugzilla, mcp_log
 
 # The FastMCP instance
 mcp = FastMCP("Bugzilla")
 
 # Global dict to hold command-line arguments, populated by main() in __init__.py
-cli_args: dict[str, Any] = {}
+cli_args: Namespace
 
 # Global variable to hold the base_url, set by the start() function
 base_url: str = ""
@@ -35,7 +33,7 @@ async def get_bz(headers: dict = CurrentHeaders()) -> Bugzilla:
     """Dependency to get the current Bugzilla client"""
     mcp_log.debug("api_key: Checking")
 
-    api_key_header = cli_args.get("api_key_header", "ApiKey")
+    api_key_header = getattr(cli_args, "api_key_header", "ApiKey")
     api_key_value = headers.get(api_key_header.lower())
 
     if not api_key_value:
@@ -45,16 +43,14 @@ async def get_bz(headers: dict = CurrentHeaders()) -> Bugzilla:
     bz = Bugzilla(
         url=base_url,
         api_key=api_key_value,
-        use_auth_header=cli_args.get("use_auth_header", False)
+        use_auth_header=getattr(cli_args, "use_auth_header", False),
     )
     try:
         yield bz
     finally:
         await bz.close()
 
 
-
-
 @mcp.tool()
 async def bug_info(id: int, bz: Bugzilla = Depends(get_bz)) -> dict[str, Any]:
     """Returns the entire information about a given bugzilla bug id"""
@@ -71,7 +67,10 @@ async def bug_info(id: int, bz: Bugzilla = Depends(get_bz)) -> dict[str, Any]:
 
 @mcp.tool()
 async def bug_comments(
-    id: int, include_private_comments: bool = False, new_since: Optional[datetime] = None, bz: Bugzilla = Depends(get_bz)
+    id: int,
+    include_private_comments: bool = False,
+    new_since: Optional[datetime] = None,
+    bz: Bugzilla = Depends(get_bz),
 ) -> List[dict[str, Any]]:
     """Returns the comments of given bug id
     Private comments are not included by default
@@ -100,7 +99,7 @@ async def bug_comments(
         raise ToolError(f"Failed to fetch bug comments\nReason: {e}")
 
 
-@mcp.tool()
+@mcp.tool(tags={"write"})
 async def add_comment(
     bug_id: int, comment: str, is_private: bool = False, bz: Bugzilla = Depends(get_bz)
 ) -> dict[str, int]:
@@ -127,7 +126,7 @@ async def bugs_quicksearch(
     bz: Bugzilla = Depends(get_bz),
 ) -> List[Any]:
     """Search bugs using bugzilla's quicksearch syntax
-    
+
     To reduce the token limit & response time, only returns a subset of fields for each bug
     The user can query full details of each bug using the bug_info tool
     """
@@ -190,7 +189,7 @@ def bug_url(bug_id: int) -> str:
 def mcp_server_info_resource() -> dict[str, Any]:
     """Returns the args being used by the current server instance"""
     mcp_log.info("[LLM-REQ] mcp_server_info_resource()")
-    info = cli_args.copy()
+    info = vars(cli_args).copy()
     info["version"] = importlib.metadata.version("mcp-bugzilla")
     return info
 
@@ -218,7 +217,7 @@ async def summarize_bug_prompt(id: int, bz: Bugzilla = Depends(get_bz)) -> str:
     - Mention usernames & dates wherever relevant.
     - date field must be in human readable format
     - Usernames must be bold italic (***username***) dates must be bold (**date**)
-    
+
     Comments Data:
     {comments}
     """.strip()
@@ -230,13 +229,13 @@ async def summarize_bug_prompt(id: int, bz: Bugzilla = Depends(get_bz)) -> str:
         raise PromptError(f"Summarize Comments Failed\nReason: {e}")
 
 
-@mcp.tool()
+@mcp.tool(tags={"write"})
 async def update_bug_status(
     bug_id: int,
     status: str,
     resolution: Optional[str] = None,
     comment: str = "",
-    bz: Bugzilla = Depends(get_bz)
+    bz: Bugzilla = Depends(get_bz),
 ) -> dict[str, Any]:
     """Update the status of a bug. Optionally add a comment explaining the status change.
 
@@ -259,7 +258,9 @@ async def update_bug_status(
 
     # Validate: CLOSED requires resolution
     if status == "CLOSED" and not resolution:
-        raise ToolError("Resolution is required when setting status to CLOSED (e.g., FIXED, WONTFIX, NOTABUG, DUPLICATE)")
+        raise ToolError(
+            "Resolution is required when setting status to CLOSED (e.g., FIXED, WONTFIX, NOTABUG, DUPLICATE)"
+        )
 
     try:
         result = await bz.update_bug(bug_id, updates, comment)
@@ -268,12 +269,9 @@ async def update_bug_status(
         raise ToolError(f"Failed to update bug status\n{e}")
 
 
-@mcp.tool()
+@mcp.tool(tags={"write"})
 async def assign_bug(
-    bug_id: int,
-    assignee: str,
-    comment: str = "",
-    bz: Bugzilla = Depends(get_bz)
+    bug_id: int, assignee: str, comment: str = "", bz: Bugzilla = Depends(get_bz)
 ) -> dict[str, Any]:
     """Assign a bug to a user. Optionally add a comment.
 
@@ -282,24 +280,22 @@ async def assign_bug(
         assignee: Email address of the assignee
         comment: Optional comment explaining the assignment
     """
-    mcp_log.info(
-        f"[LLM-REQ] assign_bug(bug_id={bug_id}, assignee='{assignee}')"
-    )
+    mcp_log.info(f"[LLM-REQ] assign_bug(bug_id={bug_id}, assignee='{assignee}')")
     try:
         result = await bz.update_bug(bug_id, {"assigned_to": assignee}, comment)
         return result
     except Exception as e:
         raise ToolError(f"Failed to assign bug\n{e}")
 
 
-@mcp.tool()
+@mcp.tool(tags={"write"})
 async def update_bug_fields(
     bug_id: int,
     priority: Optional[str] = None,
     severity: Optional[str] = None,
     resolution: Optional[str] = None,
     comment: str = "",
-    bz: Bugzilla = Depends(get_bz)
+    bz: Bugzilla = Depends(get_bz),
 ) -> dict[str, Any]:
     """Update various bug fields. All fields are optional.
 
@@ -332,34 +328,27 @@ async def update_bug_fields(
         raise ToolError(f"Failed to update bug fields\n{e}")
 
 
-@mcp.tool()
+@mcp.tool(tags={"write"})
 async def add_cc_to_bug(
-    bug_id: int,
-    cc_email: str,
-    bz: Bugzilla = Depends(get_bz)
+    bug_id: int, cc_email: str, bz: Bugzilla = Depends(get_bz)
 ) -> dict[str, Any]:
     """Add an email address to the CC list of a bug.
 
     Args:
         bug_id: Bug ID
         cc_email: Email address to add to CC list
     """
-    mcp_log.info(
-        f"[LLM-REQ] add_cc_to_bug(bug_id={bug_id}, cc_email='{cc_email}')"
-    )
+    mcp_log.info(f"[LLM-REQ] add_cc_to_bug(bug_id={bug_id}, cc_email='{cc_email}')")
     try:
         result = await bz.update_bug(bug_id, {"cc": {"add": [cc_email]}}, "")
         return result
     except Exception as e:
         raise ToolError(f"Failed to add CC\n{e}")
 
 
-@mcp.tool()
+@mcp.tool(tags={"write"})
 async def mark_as_duplicate(
-    bug_id: int,
-    duplicate_of: int,
-    comment: str = "",
-    bz: Bugzilla = Depends(get_bz)
+    bug_id: int, duplicate_of: int, comment: str = "", bz: Bugzilla = Depends(get_bz)
 ) -> dict[str, Any]:
     """Mark a bug as a duplicate of another bug and close it.
 
@@ -375,11 +364,7 @@ async def mark_as_duplicate(
     if not comment:
         comment = f"Marking as duplicate of bug {duplicate_of}"
 
-    updates = {
-        "status": "CLOSED",
-        "resolution": "DUPLICATE",
-        "dupe_of": duplicate_of
-    }
+    updates = {"status": "CLOSED", "resolution": "DUPLICATE", "dupe_of": duplicate_of}
 
     try:
         result = await bz.update_bug(bug_id, updates, comment)
@@ -412,19 +397,31 @@ def disable_components_selectively():
             mcp.disable(keys={key})
 
 
+def disable_write_components():
+    """
+    Disable all components which alter the state of bug
+    Invoked when --read-only flag is set
+    """
+    if getattr(cli_args, "read_only", False):
+        mcp_log.info("Disabling all components which can modify bugs")
+        # disable all methods with write tags
+        mcp.disable(tags={"write"})
+
+
 def start():
     """
     Starts the FastMCP server for Bugzilla.
     """
     global base_url
-    base_url = cli_args["bugzilla_server"]
+    base_url = cli_args.bugzilla_server
     # Ensure base_url doesn't have trailing slash for consistency
     if base_url.endswith("/"):
         base_url = base_url[:-1]
 
     # Seletively disable components before running the server
     disable_components_selectively()
+    disable_write_components()
 
-    mcp_log.info(f"Starting Bugzilla MCP server on {cli_args['host']}:{cli_args['port']}")
+    mcp_log.info(f"Starting Bugzilla MCP server on {cli_args.host}:{cli_args.port}")
 
-    mcp.run(transport="http", host=cli_args["host"], port=cli_args["port"])
+    mcp.run(transport="http", host=cli_args.host, port=cli_args.port)