Infer sse transport from url

docs/clients/transports.mdx

@@ -40,7 +40,7 @@ Streamable HTTP is the recommended transport for web-based deployments, providin
 #### Overview
 
 - **Class:** `fastmcp.client.transports.StreamableHttpTransport`
-- **Inferred From:** URLs starting with `http://` or `https://` (default for HTTP URLs since v2.3.0)
+- **Inferred From:** URLs starting with `http://` or `https://` (default for HTTP URLs since v2.3.0) that do not contain `/sse/` in the path
 - **Server Compatibility:** Works with FastMCP servers running in `streamable-http` mode
 
 #### Basic Usage
@@ -62,6 +62,15 @@ async def main():
 asyncio.run(main())
 ```
 
+You can also explicitly instantiate the transport:
+
+```python
+from fastmcp.client.transports import StreamableHttpTransport
+
+transport = StreamableHttpTransport(url="https://example.com/mcp")
+client = Client(transport)
+```
+
 #### Authentication with Headers
 
 For servers requiring authentication:
@@ -88,23 +97,19 @@ Server-Sent Events (SSE) is a transport that allows servers to push data to clie
 #### Overview
 
 - **Class:** `fastmcp.client.transports.SSETransport`
-- **Inferred From:** Not automatically inferred for HTTP URLs since v2.3.0 (must be explicitly specified)
+- **Inferred From:** HTTP URLs containing `/sse/` in the path
 - **Server Compatibility:** Works with FastMCP servers running in `sse` mode
 
 #### Basic Usage
 
-Since v2.3.0, you must explicitly create an `SSETransport` for SSE connections:
+The simplest way to use SSE is to let the transport be inferred from a URL with `/sse/` in the path:
 
 ```python
 from fastmcp import Client
-from fastmcp.client.transports import SSETransport
 import asyncio
 
-# Create an SSE transport
-transport = SSETransport(url="https://example.com/sse")
-
-# Pass the transport to the client
-client = Client(transport)
+# The Client automatically uses SSETransport for URLs containing /sse/ in the path
+client = Client("https://example.com/sse")
 
 async def main():
     async with client:
@@ -114,6 +119,15 @@ async def main():
 asyncio.run(main())
 ```
 
+You can also explicitly instantiate the transport for URLs that do not contain `/sse/` in the path or for more control: 
+
+```python
+from fastmcp.client.transports import SSETransport
+
+transport = SSETransport(url="https://example.com/sse")
+client = Client(transport)
+```
+
 #### Authentication with Headers
 
 SSE transport also supports custom headers for authentication:

src/fastmcp/client/transports.py

@@ -1,14 +1,13 @@
 import abc
 import contextlib
 import datetime
-import inspect
 import os
 import shutil
 import sys
-import warnings
 from collections.abc import AsyncIterator
 from pathlib import Path
 from typing import Any, TypedDict, cast
+from urllib.parse import urlparse
 
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.session import (
@@ -26,6 +25,9 @@ from pydantic import AnyUrl
 from typing_extensions import Unpack
 
 from fastmcp.server import FastMCP as FastMCPServer
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
 
 
 class SessionKwargs(TypedDict, total=False):
@@ -486,36 +488,29 @@ def infer_transport(
 
     # the transport is a FastMCP server
     elif isinstance(transport, FastMCPServer):
-        return FastMCPTransport(mcp=transport)
+        inferred_transport = FastMCPTransport(mcp=transport)
 
     # the transport is a path to a script
     elif isinstance(transport, Path | str) and Path(transport).exists():
         if str(transport).endswith(".py"):
-            return PythonStdioTransport(script_path=transport)
+            inferred_transport = PythonStdioTransport(script_path=transport)
         elif str(transport).endswith(".js"):
-            return NodeStdioTransport(script_path=transport)
+            inferred_transport = NodeStdioTransport(script_path=transport)
         else:
             raise ValueError(f"Unsupported script type: {transport}")
 
     # the transport is an http(s) URL
     elif isinstance(transport, AnyUrl | str) and str(transport).startswith("http"):
-        if str(transport).rstrip("/").endswith("/sse"):
-            warnings.warn(
-                inspect.cleandoc(
-                    """
-                    As of FastMCP 2.3.0, HTTP URLs are inferred to use Streamable HTTP.
-                    The provided URL ends in `/sse`, so you may encounter unexpected behavior.
-                    If you intended to use SSE, please use the `SSETransport` class directly.
-                    """
-                ),
-                category=UserWarning,
-                stacklevel=2,
-            )
-        return StreamableHttpTransport(url=transport)
-
-    # the transport is a websocket URL
-    elif isinstance(transport, AnyUrl | str) and str(transport).startswith("ws"):
-        return WSTransport(url=transport)
+        transport_str = str(transport)
+        # Parse out just the path portion to check for /sse
+        parsed_url = urlparse(transport_str)
+        path = parsed_url.path
+
+        # Check if path contains /sse/ or ends with /sse
+        if "/sse/" in path or path.rstrip("/").endswith("/sse"):
+            inferred_transport = SSETransport(url=transport)
+        else:
+            inferred_transport = StreamableHttpTransport(url=transport)
 
     ## if the transport is a config dict
     elif isinstance(transport, dict):
@@ -530,7 +525,7 @@ def infer_transport(
             server_name = list(server.keys())[0]
             # Stdio transport
             if "command" in server[server_name] and "args" in server[server_name]:
-                return StdioTransport(
+                inferred_transport = StdioTransport(
                     command=server[server_name]["command"],
                     args=server[server_name]["args"],
                     env=server[server_name].get("env", None),
@@ -539,7 +534,7 @@ def infer_transport(
 
             # HTTP transport
             elif "url" in server:
-                return SSETransport(
+                inferred_transport = SSETransport(
                     url=server["url"],
                     headers=server.get("headers", None),
                 )
@@ -549,3 +544,6 @@ def infer_transport(
     # the transport is an unknown type
     else:
         raise ValueError(f"Could not infer a valid transport from: {transport}")
+
+    logger.debug(f"Inferred transport: {inferred_transport}")
+    return inferred_transport

tests/client/test_client.py

@@ -6,7 +6,12 @@ from mcp import McpError
 from pydantic import AnyUrl
 
 from fastmcp.client import Client
-from fastmcp.client.transports import FastMCPTransport
+from fastmcp.client.transports import (
+    FastMCPTransport,
+    SSETransport,
+    StreamableHttpTransport,
+    infer_transport,
+)
 from fastmcp.exceptions import ResourceError, ToolError
 from fastmcp.prompts.prompt import TextContent
 from fastmcp.server.server import FastMCP
@@ -543,3 +548,53 @@ class TestTimeout:
             timeout=0.01,
         ) as client:
             await client.call_tool("sleep", {"seconds": 0.1}, timeout=2)
+
+
+class TestInferTransport:
+    """Tests for the infer_transport function."""
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "http://example.com/api/sse/stream",
+            "https://localhost:8080/mcp/sse/endpoint",
+            "http://example.com/api/sse",
+            "https://localhost:8080/mcp/sse",
+            "http://example.com/api/sse?param=value",
+            "https://localhost:8080/mcp/sse/?param=value",
+            "https://localhost:8000/mcp/sse?x=1&y=2",
+        ],
+        ids=[
+            "path_with_sse_directory",
+            "path_with_sse_subdirectory",
+            "path_ending_with_sse",
+            "path_ending_with_sse_https",
+            "path_with_sse_and_query_params",
+            "path_with_sse_slash_and_query_params",
+            "path_with_sse_and_ampersand_param",
+        ],
+    )
+    def test_url_returns_sse_transport(self, url):
+        """Test that URLs with /sse/ pattern return SSETransport."""
+        assert isinstance(infer_transport(url), SSETransport)
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "http://example.com/api",
+            "https://localhost:8080/mcp",
+            "http://example.com/asset/image.jpg",
+            "https://localhost:8080/sservice/endpoint",
+            "https://example.com/assets/file",
+        ],
+        ids=[
+            "regular_http_url",
+            "regular_https_url",
+            "url_with_unrelated_path",
+            "url_with_sservice_in_path",
+            "url_with_assets_in_path",
+        ],
+    )
+    def test_url_returns_streamable_http_transport(self, url):
+        """Test that URLs without /sse/ pattern return StreamableHttpTransport."""
+        assert isinstance(infer_transport(url), StreamableHttpTransport)