diff --git a/docs/clients/transports.mdx b/docs/clients/transports.mdx index 2b2b24ec5..4653c7233 100644 --- a/docs/clients/transports.mdx +++ b/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: diff --git a/src/fastmcp/client/transports.py b/src/fastmcp/client/transports.py index 7faeab613..debc2629d 100644 --- a/src/fastmcp/client/transports.py +++ b/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) + transport_str = str(transport) + # Parse out just the path portion to check for /sse + parsed_url = urlparse(transport_str) + path = parsed_url.path - # the transport is a websocket URL - elif isinstance(transport, AnyUrl | str) and str(transport).startswith("ws"): - return WSTransport(url=transport) + # 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 diff --git a/tests/client/test_client.py b/tests/client/test_client.py index 52d4448f8..7de6f13d0 100644 --- a/tests/client/test_client.py +++ b/tests/client/test_client.py @@ -7,7 +7,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 @@ -540,6 +545,10 @@ class TestTimeout: with pytest.raises(McpError): await client.call_tool("sleep", {"seconds": 0.1}, timeout=0.01) + @pytest.mark.skipif( + sys.platform == "win32", + reason="This test is flaky on Windows. Sometimes the client timeout is respected and sometimes it is not.", + ) async def test_timeout_tool_call_overrides_client_timeout_even_if_lower( self, fastmcp_server: FastMCP ): @@ -548,3 +557,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)