Fix MCP server mounted path

.changeset/shiny-zebras-swim.md

@@ -0,0 +1,5 @@
+---
+"gradio": patch
+---
+
+fix:Fix MCP server mounted path

gradio/mcp.py

@@ -12,7 +12,7 @@
 from mcp.server.sse import SseServerTransport
 from PIL import Image
 from starlette.applications import Starlette
-from starlette.responses import JSONResponse
+from starlette.responses import JSONResponse, Response
 from starlette.routing import Mount, Route
 
 from gradio import processing_utils, route_utils, utils
@@ -138,7 +138,7 @@ def launch_mcp_on_sse(self, app: Starlette, subpath: str, root_path: str) -> Non
             app: The Gradio app to mount the MCP server on.
             subpath: The subpath to mount the MCP server on. E.g. "/gradio_api/mcp"
         """
-        messages_path = f"{subpath}/messages/"
+        messages_path = "/messages/"
         sse = SseServerTransport(messages_path)
 
         async def handle_sse(request):
@@ -157,6 +157,7 @@ async def handle_sse(request):
                         streams[1],
                         self.mcp_server.create_initialization_options(),
                     )
+                return Response()
             except Exception as e:
                 print(f"MCP SSE connection error: {str(e)}")
                 raise

requirements-mcp.txt

@@ -1,2 +1,2 @@
-mcp>=1.6.0,<2.0.0
+mcp>=1.9.0,<2.0.0
 pydantic>=2.11; sys.platform != 'emscripten'

test/test_mcp.py

@@ -1,6 +1,7 @@
 import os
 import tempfile
 
+import httpx
 import pytest
 from PIL import Image
 
@@ -114,3 +115,42 @@ def test_tool_prefix_character_replacement():
             os.environ["SPACE_ID"] = original_space_id
         else:
             os.environ.pop("SPACE_ID", None)
+
+
+def test_mcp_sse_transport():
+    _, url, _ = app.launch(mcp_server=True, prevent_thread_lock=True)
+
+    with httpx.Client(timeout=5) as client:
+        sse_url = f"{url}gradio_api/mcp/sse"
+
+        with client.stream("GET", sse_url) as response:
+            assert response.is_success
+
+            terminate_next = False
+            line = ""
+            for line in response.iter_lines():
+                if terminate_next:
+                    break
+                if line.startswith("event: endpoint"):
+                    terminate_next = True
+
+            messages_path = line[5:].strip()
+            messages_url = f"{url.rstrip('/')}{messages_path}"
+
+            message_response = client.post(
+                messages_url,
+                json={
+                    "method": "initialize",
+                    "params": {
+                        "protocolVersion": "2025-03-26",
+                        "capabilities": {},
+                    },
+                    "jsonrpc": "2.0",
+                    "id": 0,
+                },
+                headers={"Content-Type": "application/json"},
+            )
+
+            assert message_response.is_success, (
+                f"Failed with status {message_response.status_code}: {message_response.text}"
+            )