add tool filtering capability

Author: shahar4499

examples/filtered_tools_example.py

@@ -0,0 +1,72 @@
+from examples.shared.apps import items
+from examples.shared.setup import setup_logging
+
+from fastapi_mcp import FastApiMCP
+
+setup_logging()
+
+# Example demonstrating how to filter MCP tools by operation IDs and tags
+
+# Filter by including specific operation IDs
+include_operations_mcp = FastApiMCP(
+    items.app,
+    name="Item API MCP - Included Operations",
+    description="MCP server showing only specific operations",
+    base_url="http://localhost:8001",
+    include_operations=["get_item", "list_items"],
+)
+
+# Filter by excluding specific operation IDs
+exclude_operations_mcp = FastApiMCP(
+    items.app,
+    name="Item API MCP - Excluded Operations",
+    description="MCP server showing all operations except the excluded ones",
+    base_url="http://localhost:8002",
+    exclude_operations=["create_item", "update_item", "delete_item"],
+)
+
+# Filter by including specific tags
+include_tags_mcp = FastApiMCP(
+    items.app,
+    name="Item API MCP - Included Tags",
+    description="MCP server showing operations with specific tags",
+    base_url="http://localhost:8003",
+    include_tags=["items"],
+)
+
+# Filter by excluding specific tags
+exclude_tags_mcp = FastApiMCP(
+    items.app,
+    name="Item API MCP - Excluded Tags",
+    description="MCP server showing operations except those with specific tags",
+    base_url="http://localhost:8004",
+    exclude_tags=["search"],
+)
+
+# Combine operation IDs and tags (include mode)
+combined_include_mcp = FastApiMCP(
+    items.app,
+    name="Item API MCP - Combined Include",
+    description="MCP server showing operations by combining include filters",
+    base_url="http://localhost:8005",
+    include_operations=["delete_item"],
+    include_tags=["search"],
+)
+
+# Mount all MCP servers with different paths
+include_operations_mcp.mount(mount_path="/include-operations-mcp")
+exclude_operations_mcp.mount(mount_path="/exclude-operations-mcp")
+include_tags_mcp.mount(mount_path="/include-tags-mcp")
+exclude_tags_mcp.mount(mount_path="/exclude-tags-mcp")
+combined_include_mcp.mount(mount_path="/combined-include-mcp")
+
+if __name__ == "__main__":
+    import uvicorn
+
+    print("Server is running with multiple MCP endpoints:")
+    print(" - /include-operations-mcp: Only get_item and list_items operations")
+    print(" - /exclude-operations-mcp: All operations except create_item, update_item, and delete_item")
+    print(" - /include-tags-mcp: Only operations with the 'items' tag")
+    print(" - /exclude-tags-mcp: All operations except those with the 'search' tag")
+    print(" - /combined-include-mcp: Operations with 'search' tag or delete_item operation")
+    uvicorn.run(items.app, host="0.0.0.0", port=8000)

fastapi_mcp/server.py

@@ -27,6 +27,10 @@ def __init__(
         describe_all_responses: bool = False,
         describe_full_response_schema: bool = False,
         http_client: Optional[AsyncClientProtocol] = None,
+        include_operations: Optional[List[str]] = None,
+        exclude_operations: Optional[List[str]] = None,
+        include_tags: Optional[List[str]] = None,
+        exclude_tags: Optional[List[str]] = None,
     ):
         """
         Create an MCP server from a FastAPI app.
@@ -42,7 +46,17 @@ def __init__(
             describe_full_response_schema: Whether to include full json schema for responses in tool descriptions
             http_client: Optional HTTP client to use for API calls. If not provided, a new httpx.AsyncClient will be created.
                 This is primarily for testing purposes.
+            include_operations: List of operation IDs to include as MCP tools. Cannot be used with exclude_operations.
+            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.
         """
+        # Validate operation and tag filtering options
+        if include_operations is not None and exclude_operations is not None:
+            raise ValueError("Cannot specify both include_operations and exclude_operations")
+
+        if include_tags is not None and exclude_tags is not None:
+            raise ValueError("Cannot specify both include_tags and exclude_tags")
 
         self.operation_map: Dict[str, Dict[str, Any]]
         self.tools: List[types.Tool]
@@ -55,6 +69,10 @@ def __init__(
         self._base_url = base_url
         self._describe_all_responses = describe_all_responses
         self._describe_full_response_schema = describe_full_response_schema
+        self._include_operations = include_operations
+        self._exclude_operations = exclude_operations
+        self._include_tags = include_tags
+        self._exclude_tags = exclude_tags
 
         self._http_client = http_client or httpx.AsyncClient()
 
@@ -71,12 +89,15 @@ def setup_server(self) -> None:
         )
 
         # Convert OpenAPI schema to MCP tools
-        self.tools, self.operation_map = convert_openapi_to_mcp_tools(
+        all_tools, self.operation_map = convert_openapi_to_mcp_tools(
             openapi_schema,
             describe_all_responses=self._describe_all_responses,
             describe_full_response_schema=self._describe_full_response_schema,
         )
 
+        # Filter tools based on operation IDs and tags
+        self.tools = self._filter_tools(all_tools, openapi_schema)
+
         # Determine base URL if not provided
         if not self._base_url:
             # Try to determine the base URL from FastAPI config
@@ -266,3 +287,67 @@ async def _request(
             return await client.patch(url, params=query, headers=headers, json=body)
         else:
             raise ValueError(f"Unsupported HTTP method: {method}")
+
+    def _filter_tools(self, tools: List[types.Tool], openapi_schema: Dict[str, Any]) -> List[types.Tool]:
+        """
+        Filter tools based on operation IDs and tags.
+
+        Args:
+            tools: List of tools to filter
+            openapi_schema: The OpenAPI schema
+
+        Returns:
+            Filtered list of tools
+        """
+        if (
+            self._include_operations is None
+            and self._exclude_operations is None
+            and self._include_tags is None
+            and self._exclude_tags is None
+        ):
+            return tools
+
+        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"]:
+                    continue
+
+                operation_id = operation.get("operationId")
+                if not operation_id:
+                    continue
+
+                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)
+
+        operations_to_include = set()
+
+        if self._include_operations is not None:
+            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))
+
+        if self._include_tags is not None:
+            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 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
+            }
+
+        return filtered_tools

tests/test_configuration.py

@@ -1,4 +1,5 @@
 from fastapi import FastAPI
+import pytest
 
 from fastapi_mcp import FastApiMCP
 
@@ -374,3 +375,65 @@ def test_describe_all_responses_and_full_response_schema_config_complex_app(comp
             assert tool.description.count("**Output Schema:**") > 0, (
                 "The description should contain full output schemas"
             )
+
+
+def test_filtering_functionality():
+    """Test that FastApiMCP correctly filters endpoints based on operation IDs and tags."""
+    app = FastAPI()
+
+    # Define endpoints with different operation IDs and tags
+    @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}
+
+    @app.get("/search/", operation_id="search_items", tags=["search"])
+    async def search_items():
+        return [{"id": 1}]
+
+    # Test include_operations
+    include_ops_mcp = FastApiMCP(app, include_operations=["get_item", "list_items"])
+    assert len(include_ops_mcp.tools) == 2
+    assert {tool.name for tool in include_ops_mcp.tools} == {"get_item", "list_items"}
+
+    # Test exclude_operations
+    exclude_ops_mcp = FastApiMCP(app, exclude_operations=["delete_item", "search_items"])
+    assert len(exclude_ops_mcp.tools) == 4
+    assert {tool.name for tool in exclude_ops_mcp.tools} == {"get_item", "list_items", "create_item", "update_item"}
+
+    # Test include_tags
+    include_tags_mcp = FastApiMCP(app, include_tags=["read"])
+    assert len(include_tags_mcp.tools) == 1
+    assert {tool.name for tool in include_tags_mcp.tools} == {"get_item"}
+
+    # Test exclude_tags
+    exclude_tags_mcp = FastApiMCP(app, exclude_tags=["write", "delete"])
+    assert len(exclude_tags_mcp.tools) == 3
+    assert {tool.name for tool in exclude_tags_mcp.tools} == {"get_item", "list_items", "search_items"}
+
+    # Test combining include_operations and include_tags
+    combined_include_mcp = FastApiMCP(app, include_operations=["delete_item"], include_tags=["search"])
+    assert len(combined_include_mcp.tools) == 2
+    assert {tool.name for tool in combined_include_mcp.tools} == {"delete_item", "search_items"}
+
+    # Test invalid combinations
+    with pytest.raises(ValueError):
+        FastApiMCP(app, include_operations=["get_item"], exclude_operations=["delete_item"])
+
+    with pytest.raises(ValueError):
+        FastApiMCP(app, include_tags=["items"], exclude_tags=["write"])