fix compatibility

Author: naji247

.github/workflows/mcp-compatibility.yml

@@ -23,14 +23,15 @@ jobs:
         # Get all available versions from PyPI
         versions=$(pip index versions mcp 2>/dev/null | grep -o '[0-9]\+\.[0-9]\+\.[0-9]\+' | sort -V)
         
-        # Filter to versions >= 1.0.0 and get only latest patch version for each minor
+        # Filter to versions >= 1.2.0 and get only latest patch version for each minor
         declare -A latest_minor
         for version in $versions; do
           major=$(echo $version | cut -d. -f1)
           minor=$(echo $version | cut -d. -f2)
+          patch=$(echo $version | cut -d. -f3)
           
-          # Include if major >= 1
-          if [ "$major" -ge 1 ]; then
+          # Include if version >= 1.2.0
+          if [ "$major" -gt 1 ] || ([ "$major" -eq 1 ] && [ "$minor" -ge 2 ]); then
             minor_key="$major.$minor"
             latest_minor[$minor_key]="$version"
           fi

src/mcpcat/modules/overrides/fastmcp.py

@@ -1,7 +1,6 @@
-from typing import Any
+from typing import Any, TYPE_CHECKING
 
 from mcp import ServerResult, Tool
-from mcp.server import FastMCP
 from mcp.types import CallToolRequest, CallToolResult, ListToolsRequest, TextContent
 
 from mcpcat.modules.overrides.mcp_server import override_lowlevel_mcp_server
@@ -11,11 +10,22 @@
 from ...types import MCPCatData
 from ..logging import log_info, log_warning
 from ..session import capture_session_info
+from ..version_detection import has_fastmcp_support
 
 """Tool management and interception for MCPCat."""
 
+if TYPE_CHECKING or has_fastmcp_support():
+    try:
+        from mcp.server import FastMCP
+    except ImportError:
+        FastMCP = None
+else:
+    FastMCP = None
 
-def override_fastmcp(server: FastMCP, data: MCPCatData) -> None:
+
+def override_fastmcp(server: Any, data: MCPCatData) -> None:
     """Set up tool list and call handlers for FastMCP."""
+    if FastMCP is None:
+        raise ImportError("FastMCP is not available in this MCP version")
     from mcp.types import CallToolResult, ListToolsResult
     override_lowlevel_mcp_server(server._mcp_server, data)

src/mcpcat/modules/overrides/mcp_server.py

@@ -1,7 +1,7 @@
 from typing import Any
 
 from mcp import ServerResult, Tool
-from mcp.server.lowlevel import Server
+from mcp.server import Server
 from mcp.types import CallToolRequest, CallToolResult, ListToolsRequest, TextContent
 
 from mcpcat.modules.tools import handle_report_missing

src/mcpcat/modules/tools.py

@@ -1,17 +1,23 @@
 """Tool management and interception for MCPCat."""
 
-from typing import Any
+from typing import Any, TYPE_CHECKING
 
 from mcp import ServerResult, Tool
-from mcp.server import FastMCP
 from mcp.types import CallToolRequest, CallToolResult, ListToolsRequest, TextContent
 
 from mcpcat.modules.tracing import record_trace
+from mcpcat.modules.version_detection import has_fastmcp_support
 
 from ..types import MCPCatData
 from .logging import log_info
 from .session import capture_session_info
 
+if TYPE_CHECKING or has_fastmcp_support():
+    try:
+        from mcp.server import FastMCP
+    except ImportError:
+        FastMCP = None
+
 async def handle_report_missing(arguments: dict[str, Any], data: MCPCatData) -> CallToolResult:
     """Handle the report_missing tool."""
     missing_tool = arguments.get("missing_tool", "")

src/mcpcat/modules/version_detection.py

@@ -0,0 +1,47 @@
+"""MCP version detection utilities."""
+
+import importlib.metadata
+from typing import Tuple, Optional
+
+
+def get_mcp_version() -> Optional[str]:
+    """Get the installed MCP version."""
+    try:
+        return importlib.metadata.version("mcp")
+    except importlib.metadata.PackageNotFoundError:
+        return None
+
+
+def parse_version(version_str: str) -> Tuple[int, int, int]:
+    """Parse version string to tuple of integers."""
+    parts = version_str.split(".")
+    major = int(parts[0]) if len(parts) > 0 else 0
+    minor = int(parts[1]) if len(parts) > 1 else 0
+    patch = int(parts[2]) if len(parts) > 2 else 0
+    return (major, minor, patch)
+
+
+def has_fastmcp_support() -> bool:
+    """Check if the current MCP version supports FastMCP."""
+    version = get_mcp_version()
+    if not version:
+        return False
+    
+    major, minor, _ = parse_version(version)
+    
+    # FastMCP was introduced after version 1.1
+    if major < 1:
+        return False
+    if major == 1 and minor <= 1:
+        return False
+    
+    return True
+
+
+def can_import_fastmcp() -> bool:
+    """Check if FastMCP can be imported."""
+    try:
+        from mcp.server import FastMCP
+        return True
+    except ImportError:
+        return False

tests/test_mcp_version_compatibility_comprehensive.py

@@ -3,12 +3,18 @@
 from contextlib import asynccontextmanager
 from typing import Any
 
-from mcp.server import FastMCP
 from mcp.shared.memory import create_connected_server_and_client_session
 
+try:
+    from mcp.server import FastMCP
+    HAS_FASTMCP = True
+except ImportError:
+    FastMCP = None
+    HAS_FASTMCP = False
+
 
 @asynccontextmanager
-async def create_test_client(server: FastMCP):
+async def create_test_client(server: Any):
     """Create a test client for the given server.
     
     This creates a properly connected MCP client/server pair with full
@@ -21,11 +27,18 @@ async def create_test_client(server: FastMCP):
         async with create_test_client(server) as client:
             result = await client.call_tool("add_todo", {"text": "Test"})
     """
-    async with create_connected_server_and_client_session(server._mcp_server) as client:
-        yield client
+    # Handle both FastMCP and low-level Server
+    if hasattr(server, '_mcp_server'):
+        # FastMCP server
+        async with create_connected_server_and_client_session(server._mcp_server) as client:
+            yield client
+    else:
+        # Low-level Server
+        async with create_connected_server_and_client_session(server) as client:
+            yield client
 
 
-async def call_tool_via_client(server: FastMCP, tool_name: str, arguments: dict[str, Any]) -> Any:
+async def call_tool_via_client(server: Any, tool_name: str, arguments: dict[str, Any]) -> Any:
     """Helper to call a tool through a proper client session."""
     async with create_test_client(server) as client:
         result = await client.call_tool(tool_name, arguments)

tests/test_utils/client.py

@@ -1,7 +1,13 @@
 """Todo server implementation for testing."""
 
+from mcp.server import Server
 
-from mcp.server import FastMCP, Server
+try:
+    from mcp.server import FastMCP
+    HAS_FASTMCP = True
+except ImportError:
+    FastMCP = None
+    HAS_FASTMCP = False
 
 
 class Todo:
@@ -13,8 +19,10 @@ def __init__(self, id: str, text: str, completed: bool = False):
         self.completed = completed
 
 
-def create_todo_server() -> FastMCP:
+def create_todo_server():
     """Create a todo server for testing."""
+    if FastMCP is None:
+        raise ImportError("FastMCP is not available in this MCP version. Use create_low_level_todo_server() instead.")
     # Fix deprecation warning by not passing version as kwarg
     server = FastMCP("todo-server")
 
@@ -61,99 +69,3 @@ def complete_todo(id: str) -> str:
     }
 
     return server
-
-
-def create_low_level_todo_server() -> Server:
-    """Create a low-level MCP server for testing."""
-    from mcp.types import Tool, TextContent
-    
-    server = Server("todo-server-lowlevel")
-    
-    todos: list[Todo] = []
-    next_id = 1
-    
-    # Define tools
-    add_todo_tool = Tool(
-        name="add_todo",
-        description="Add a new todo item",
-        inputSchema={
-            "type": "object",
-            "properties": {
-                "text": {
-                    "type": "string",
-                    "description": "The todo item text"
-                }
-            },
-            "required": ["text"]
-        }
-    )
-    
-    list_todos_tool = Tool(
-        name="list_todos", 
-        description="List all todo items",
-        inputSchema={
-            "type": "object",
-            "properties": {}
-        }
-    )
-    
-    complete_todo_tool = Tool(
-        name="complete_todo",
-        description="Mark a todo item as completed",
-        inputSchema={
-            "type": "object", 
-            "properties": {
-                "id": {
-                    "type": "string",
-                    "description": "The todo item ID"
-                }
-            },
-            "required": ["id"]
-        }
-    )
-    
-    @server.list_tools()
-    async def list_tools():
-        return [add_todo_tool, list_todos_tool, complete_todo_tool]
-    
-    @server.call_tool()
-    async def call_tool(name: str, arguments: dict):
-        nonlocal next_id
-        
-        if name == "add_todo":
-            text = arguments["text"]
-            todo = Todo(str(next_id), text)
-            todos.append(todo)
-            next_id += 1
-            return [TextContent(
-                type="text",
-                text=f'Added todo: "{text}" with ID {todo.id}'
-            )]
-        
-        elif name == "list_todos":
-            if not todos:
-                return [TextContent(type="text", text="No todos found")]
-            
-            todo_list = []
-            for todo in todos:
-                status = "✓" if todo.completed else "○"
-                todo_list.append(f"{todo.id}: {todo.text} {status}")
-            
-            return [TextContent(type="text", text="\n".join(todo_list))]
-        
-        elif name == "complete_todo":
-            todo_id = arguments["id"]
-            for todo in todos:
-                if todo.id == todo_id:
-                    todo.completed = True
-                    return [TextContent(
-                        type="text",
-                        text=f'Completed todo: "{todo.text}"'
-                    )]
-            
-            raise ValueError(f"Todo with ID {todo_id} not found")
-        
-        else:
-            raise ValueError(f"Unknown tool: {name}")
-    
-    return server