Filter get endpoints

fastapi_mcp/server.py

@@ -31,6 +31,7 @@ def __init__(
         exclude_operations: Optional[List[str]] = None,
         include_tags: Optional[List[str]] = None,
         exclude_tags: Optional[List[str]] = None,
+        only_get_endpoints: bool = False,
     ):
         """
         Create an MCP server from a FastAPI app.
@@ -50,6 +51,7 @@ def __init__(
             exclude_operations: List of operation IDs to exclude from MCP tools. Cannot be used with include_operations.
             include_tags: List of tags to include as MCP tools. Cannot be used with exclude_tags.
             exclude_tags: List of tags to exclude from MCP tools. Cannot be used with include_tags.
+            only_get_endpoints: If True, only expose GET endpoints. This filter is applied after other filters.
         """
         # Validate operation and tag filtering options
         if include_operations is not None and exclude_operations is not None:
@@ -73,6 +75,7 @@ def __init__(
         self._exclude_operations = exclude_operations
         self._include_tags = include_tags
         self._exclude_tags = exclude_tags
+        self._only_get_endpoints = only_get_endpoints
 
         self._http_client = http_client or httpx.AsyncClient()
 
@@ -316,15 +319,20 @@ def _filter_tools(self, tools: List[types.Tool], openapi_schema: Dict[str, Any])
         Returns:
             Filtered list of tools
         """
+        # Early return if no filters are applied
         if (
             self._include_operations is None
             and self._exclude_operations is None
             and self._include_tags is None
             and self._exclude_tags is None
+            and not self._only_get_endpoints
         ):
             return tools
 
+        # Build mapping of operation IDs to their HTTP methods
+        operation_methods: Dict[str, str] = {}
         operations_by_tag: Dict[str, List[str]] = {}
+
         for path, path_item in openapi_schema.get("paths", {}).items():
             for method, operation in path_item.items():
                 if method not in ["get", "post", "put", "delete", "patch"]:
@@ -333,38 +341,58 @@ def _filter_tools(self, tools: List[types.Tool], openapi_schema: Dict[str, Any])
                 operation_id = operation.get("operationId")
                 if not operation_id:
                     continue
+
+                # Store the HTTP method for each operation ID
+                operation_methods[operation_id] = method
 
                 tags = operation.get("tags", [])
                 for tag in tags:
                     if tag not in operations_by_tag:
                         operations_by_tag[tag] = []
                     operations_by_tag[tag].append(operation_id)
 
+        # Get all tool operation IDs
+        all_operations = {tool.name for tool in tools}
         operations_to_include = set()
 
+        # Handle empty include lists specially - they should result in no tools
         if self._include_operations is not None:
+            if not self._include_operations:  # Empty list means include nothing
+                return []
             operations_to_include.update(self._include_operations)
         elif self._exclude_operations is not None:
-            all_operations = {tool.name for tool in tools}
             operations_to_include.update(all_operations - set(self._exclude_operations))
 
+        # Apply tag filters
         if self._include_tags is not None:
+            if not self._include_tags:  # Empty list means include nothing
+                return []
             for tag in self._include_tags:
                 operations_to_include.update(operations_by_tag.get(tag, []))
         elif self._exclude_tags is not None:
             excluded_operations = set()
             for tag in self._exclude_tags:
                 excluded_operations.update(operations_by_tag.get(tag, []))
-
-            all_operations = {tool.name for tool in tools}
             operations_to_include.update(all_operations - excluded_operations)
 
-        filtered_tools = [tool for tool in tools if tool.name in operations_to_include]
+        # If no filters applied yet (but only_get_endpoints is True), include all operations
+        if not operations_to_include and self._only_get_endpoints:
+            operations_to_include = all_operations
 
+        # Apply GET-only filter if enabled
+        if self._only_get_endpoints:
+            get_operations = {op_id for op_id, method in operation_methods.items() if method.lower() == "get"}
+            operations_to_include &= get_operations  # Use set intersection operator
+
+        # Filter tools based on the final set of operations to include
+        filtered_tools = [tool for tool in tools if tool.name in operations_to_include]
+
+        # Update operation_map with only the filtered operations
         if filtered_tools:
             filtered_operation_ids = {tool.name for tool in filtered_tools}
             self.operation_map = {
-                op_id: details for op_id, details in self.operation_map.items() if op_id in filtered_operation_ids
+                op_id: details for op_id, details in self.operation_map.items() 
+                if op_id in filtered_operation_ids
             }
 
         return filtered_tools

tests/test_configuration.py

@@ -615,3 +615,77 @@ async def empty_tags():
     exclude_tags_mcp = FastApiMCP(app, exclude_tags=["items"])
     assert len(exclude_tags_mcp.tools) == 1
     assert {tool.name for tool in exclude_tags_mcp.tools} == {"empty_tags"}
+
+
+def test_only_get_endpoints_filtering():
+    """Test that FastApiMCP correctly filters to only GET endpoints when only_get_endpoints is True."""
+    app = FastAPI()
+
+    # Define endpoints with different HTTP methods
+    @app.get("/items/", operation_id="list_items", tags=["items"])
+    async def list_items():
+        return [{"id": 1}]
+
+    @app.get("/items/{item_id}", operation_id="get_item", tags=["items", "read"])
+    async def get_item(item_id: int):
+        return {"id": item_id}
+
+    @app.post("/items/", operation_id="create_item", tags=["items", "write"])
+    async def create_item():
+        return {"id": 2}
+
+    @app.put("/items/{item_id}", operation_id="update_item", tags=["items", "write"])
+    async def update_item(item_id: int):
+        return {"id": item_id}
+
+    @app.delete("/items/{item_id}", operation_id="delete_item", tags=["items", "delete"])
+    async def delete_item(item_id: int):
+        return {"id": item_id}
+
+    # Test only_get_endpoints=True
+    only_get_mcp = FastApiMCP(app, only_get_endpoints=True)
+    assert len(only_get_mcp.tools) == 2
+    assert {tool.name for tool in only_get_mcp.tools} == {"get_item", "list_items"}
+
+    # Test only_get_endpoints with include_operations
+    get_with_include_ops = FastApiMCP(
+        app, 
+        only_get_endpoints=True, 
+        include_operations=["get_item", "create_item", "update_item"]
+    )
+    assert len(get_with_include_ops.tools) == 1
+    assert {tool.name for tool in get_with_include_ops.tools} == {"get_item"}
+
+    # Test only_get_endpoints with exclude_operations
+    get_with_exclude_ops = FastApiMCP(
+        app, 
+        only_get_endpoints=True, 
+        exclude_operations=["list_items"]
+    )
+    assert len(get_with_exclude_ops.tools) == 1
+    assert {tool.name for tool in get_with_exclude_ops.tools} == {"get_item"}
+
+    # Test only_get_endpoints with include_tags
+    get_with_include_tags = FastApiMCP(
+        app, 
+        only_get_endpoints=True, 
+        include_tags=["read"]
+    )
+    assert len(get_with_include_tags.tools) == 1
+    assert {tool.name for tool in get_with_include_tags.tools} == {"get_item"}
+
+    # Test only_get_endpoints with exclude_tags
+    get_with_exclude_tags = FastApiMCP(
+        app, 
+        only_get_endpoints=True, 
+        exclude_tags=["read"]
+    )
+    assert len(get_with_exclude_tags.tools) == 1
+    assert {tool.name for tool in get_with_exclude_tags.tools} == {"list_items"}
+
+    # Test only_get_endpoints=False (should include all endpoints)
+    all_methods_mcp = FastApiMCP(app, only_get_endpoints=False)
+    assert len(all_methods_mcp.tools) == 5
+    assert {tool.name for tool in all_methods_mcp.tools} == {
+        "get_item", "list_items", "create_item", "update_item", "delete_item"
+    }