From 219e1f6bd1dfdb523b8108bed30cc688cf8f2d15 Mon Sep 17 00:00:00 2001 From: Itay Shemer Date: Tue, 15 Apr 2025 09:55:49 +0300 Subject: [PATCH 1/2] add GET filter --- fastapi_mcp/server.py | 38 +++++++++++++++++++++++++++++++++----- 1 file changed, 33 insertions(+), 5 deletions(-) diff --git a/fastapi_mcp/server.py b/fastapi_mcp/server.py index fbff6e8..ba4f94c 100644 --- a/fastapi_mcp/server.py +++ b/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,6 +341,9 @@ 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: @@ -340,31 +351,48 @@ def _filter_tools(self, tools: List[types.Tool], openapi_schema: Dict[str, Any]) 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 From b55afdf63b474d68e93e785e62cd6756a934ec6c Mon Sep 17 00:00:00 2001 From: Itay Shemer Date: Tue, 15 Apr 2025 09:55:57 +0300 Subject: [PATCH 2/2] add tests --- tests/test_configuration.py | 74 +++++++++++++++++++++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/tests/test_configuration.py b/tests/test_configuration.py index 3b172cc..978a70e 100644 --- a/tests/test_configuration.py +++ b/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" + }