From 8b0d016c30b54e82f1695b3a1fd58d1049ac2d07 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Sun, 18 Jan 2026 22:11:25 -0500 Subject: [PATCH] Split transports.py into modular structure (#2921) --- docs/docs.json | 13 +- .../fastmcp-client-transports-__init__.mdx | 8 + .../fastmcp-client-transports-base.mdx | 62 + .../fastmcp-client-transports-config.mdx | 72 + .../fastmcp-client-transports-http.mdx | 48 + .../fastmcp-client-transports-inference.mdx | 56 + .../fastmcp-client-transports-memory.mdx | 27 + .../fastmcp-client-transports-stdio.mdx | 79 ++ docs/python-sdk/fastmcp-client-transports.mdx | 306 ----- justfile | 2 +- loq.toml | 4 - src/fastmcp/client/transports.py | 1211 ----------------- src/fastmcp/client/transports/__init__.py | 37 + src/fastmcp/client/transports/base.py | 82 ++ src/fastmcp/client/transports/config.py | 170 +++ src/fastmcp/client/transports/http.py | 213 +++ src/fastmcp/client/transports/inference.py | 153 +++ src/fastmcp/client/transports/memory.py | 90 ++ src/fastmcp/client/transports/stdio.py | 543 ++++++++ 19 files changed, 1653 insertions(+), 1523 deletions(-) create mode 100644 docs/python-sdk/fastmcp-client-transports-__init__.mdx create mode 100644 docs/python-sdk/fastmcp-client-transports-base.mdx create mode 100644 docs/python-sdk/fastmcp-client-transports-config.mdx create mode 100644 docs/python-sdk/fastmcp-client-transports-http.mdx create mode 100644 docs/python-sdk/fastmcp-client-transports-inference.mdx create mode 100644 docs/python-sdk/fastmcp-client-transports-memory.mdx create mode 100644 docs/python-sdk/fastmcp-client-transports-stdio.mdx delete mode 100644 docs/python-sdk/fastmcp-client-transports.mdx delete mode 100644 src/fastmcp/client/transports.py create mode 100644 src/fastmcp/client/transports/__init__.py create mode 100644 src/fastmcp/client/transports/base.py create mode 100644 src/fastmcp/client/transports/config.py create mode 100644 src/fastmcp/client/transports/http.py create mode 100644 src/fastmcp/client/transports/inference.py create mode 100644 src/fastmcp/client/transports/memory.py create mode 100644 src/fastmcp/client/transports/stdio.py diff --git a/docs/docs.json b/docs/docs.json index 4fcc03c3b..c27f55a00 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -354,7 +354,18 @@ }, "python-sdk/fastmcp-client-tasks", "python-sdk/fastmcp-client-telemetry", - "python-sdk/fastmcp-client-transports" + { + "group": "transports", + "pages": [ + "python-sdk/fastmcp-client-transports-__init__", + "python-sdk/fastmcp-client-transports-base", + "python-sdk/fastmcp-client-transports-config", + "python-sdk/fastmcp-client-transports-http", + "python-sdk/fastmcp-client-transports-inference", + "python-sdk/fastmcp-client-transports-memory", + "python-sdk/fastmcp-client-transports-stdio" + ] + } ] }, { diff --git a/docs/python-sdk/fastmcp-client-transports-__init__.mdx b/docs/python-sdk/fastmcp-client-transports-__init__.mdx new file mode 100644 index 000000000..1f9b02d38 --- /dev/null +++ b/docs/python-sdk/fastmcp-client-transports-__init__.mdx @@ -0,0 +1,8 @@ +--- +title: __init__ +sidebarTitle: __init__ +--- + +# `fastmcp.client.transports` + +*This module is empty or contains only private/internal implementations.* diff --git a/docs/python-sdk/fastmcp-client-transports-base.mdx b/docs/python-sdk/fastmcp-client-transports-base.mdx new file mode 100644 index 000000000..a3e6e9499 --- /dev/null +++ b/docs/python-sdk/fastmcp-client-transports-base.mdx @@ -0,0 +1,62 @@ +--- +title: base +sidebarTitle: base +--- + +# `fastmcp.client.transports.base` + +## Classes + +### `SessionKwargs` + + +Keyword arguments for the MCP ClientSession constructor. + + +### `ClientTransport` + + +Abstract base class for different MCP client transport mechanisms. + +A Transport is responsible for establishing and managing connections +to an MCP server, and providing a ClientSession within an async context. + + +**Methods:** + +#### `connect_session` + +```python +connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] +``` + +Establishes a connection and yields an active ClientSession. + +The ClientSession is *not* expected to be initialized in this context manager. + +The session is guaranteed to be valid only within the scope of the +async context manager. Connection setup and teardown are handled +within this context. + +**Args:** +- `**session_kwargs`: Keyword arguments to pass to the ClientSession + constructor (e.g., callbacks, timeouts). + + +#### `close` + +```python +close(self) +``` + +Close the transport. + + +#### `get_session_id` + +```python +get_session_id(self) -> str | None +``` + +Get the session ID for this transport, if available. + diff --git a/docs/python-sdk/fastmcp-client-transports-config.mdx b/docs/python-sdk/fastmcp-client-transports-config.mdx new file mode 100644 index 000000000..d19f709f2 --- /dev/null +++ b/docs/python-sdk/fastmcp-client-transports-config.mdx @@ -0,0 +1,72 @@ +--- +title: config +sidebarTitle: config +--- + +# `fastmcp.client.transports.config` + +## Classes + +### `MCPConfigTransport` + + +Transport for connecting to one or more MCP servers defined in an MCPConfig. + +This transport provides a unified interface to multiple MCP servers defined in an MCPConfig +object or dictionary matching the MCPConfig schema. It supports two key scenarios: + +1. If the MCPConfig contains exactly one server, it creates a direct transport to that server. +2. If the MCPConfig contains multiple servers, it creates a composite client by mounting + all servers on a single FastMCP instance, with each server's name, by default, used as its mounting prefix. + +In the multi-server case, tools are accessible with the prefix pattern `{server_name}_{tool_name}` +and resources with the pattern `protocol://{server_name}/path/to/resource`. + +This is particularly useful for creating clients that need to interact with multiple specialized +MCP servers through a single interface, simplifying client code. + +**Examples:** + +```python +from fastmcp import Client + +# Create a config with multiple servers +config = { + "mcpServers": { + "weather": { + "url": "https://weather-api.example.com/mcp", + "transport": "http" + }, + "calendar": { + "url": "https://calendar-api.example.com/mcp", + "transport": "http" + } + } +} + +# Create a client with the config +client = Client(config) + +async with client: + # Access tools with prefixes + weather = await client.call_tool("weather_get_forecast", {"city": "London"}) + events = await client.call_tool("calendar_list_events", {"date": "2023-06-01"}) + + # Access resources with prefixed URIs + icons = await client.read_resource("weather://weather/icons/sunny") +``` + + +**Methods:** + +#### `connect_session` + +```python +connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] +``` + +#### `close` + +```python +close(self) +``` diff --git a/docs/python-sdk/fastmcp-client-transports-http.mdx b/docs/python-sdk/fastmcp-client-transports-http.mdx new file mode 100644 index 000000000..584632184 --- /dev/null +++ b/docs/python-sdk/fastmcp-client-transports-http.mdx @@ -0,0 +1,48 @@ +--- +title: http +sidebarTitle: http +--- + +# `fastmcp.client.transports.http` + +## Classes + +### `SSETransport` + + +Transport implementation that connects to an MCP server via Server-Sent Events. + + +**Methods:** + +#### `connect_session` + +```python +connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] +``` + +### `StreamableHttpTransport` + + +Transport implementation that connects to an MCP server via Streamable HTTP Requests. + + +**Methods:** + +#### `connect_session` + +```python +connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] +``` + +#### `get_session_id` + +```python +get_session_id(self) -> str | None +``` + +#### `close` + +```python +close(self) +``` diff --git a/docs/python-sdk/fastmcp-client-transports-inference.mdx b/docs/python-sdk/fastmcp-client-transports-inference.mdx new file mode 100644 index 000000000..56aac235d --- /dev/null +++ b/docs/python-sdk/fastmcp-client-transports-inference.mdx @@ -0,0 +1,56 @@ +--- +title: inference +sidebarTitle: inference +--- + +# `fastmcp.client.transports.inference` + +## Functions + +### `infer_transport` + +```python +infer_transport(transport: ClientTransport | FastMCP | FastMCP1Server | AnyUrl | Path | MCPConfig | dict[str, Any] | str) -> ClientTransport +``` + + +Infer the appropriate transport type from the given transport argument. + +This function attempts to infer the correct transport type from the provided +argument, handling various input types and converting them to the appropriate +ClientTransport subclass. + +The function supports these input types: +- ClientTransport: Used directly without modification +- FastMCP or FastMCP1Server: Creates an in-memory FastMCPTransport +- Path or str (file path): Creates PythonStdioTransport (.py) or NodeStdioTransport (.js) +- AnyUrl or str (URL): Creates StreamableHttpTransport (default) or SSETransport (for /sse endpoints) +- MCPConfig or dict: Creates MCPConfigTransport, potentially connecting to multiple servers + +For HTTP URLs, they are assumed to be Streamable HTTP URLs unless they end in `/sse`. + +For MCPConfig with multiple servers, a composite client is created where each server +is mounted with its name as prefix. This allows accessing tools and resources from multiple +servers through a single unified client interface, using naming patterns like +`servername_toolname` for tools and `protocol://servername/path` for resources. +If the MCPConfig contains only one server, a direct connection is established without prefixing. + +**Examples:** + +```python +# Connect to a local Python script +transport = infer_transport("my_script.py") + +# Connect to a remote server via HTTP +transport = infer_transport("http://example.com/mcp") + +# Connect to multiple servers using MCPConfig +config = { + "mcpServers": { + "weather": {"url": "http://weather.example.com/mcp"}, + "calendar": {"url": "http://calendar.example.com/mcp"} + } +} +transport = infer_transport(config) +``` + diff --git a/docs/python-sdk/fastmcp-client-transports-memory.mdx b/docs/python-sdk/fastmcp-client-transports-memory.mdx new file mode 100644 index 000000000..6d9999ec9 --- /dev/null +++ b/docs/python-sdk/fastmcp-client-transports-memory.mdx @@ -0,0 +1,27 @@ +--- +title: memory +sidebarTitle: memory +--- + +# `fastmcp.client.transports.memory` + +## Classes + +### `FastMCPTransport` + + +In-memory transport for FastMCP servers. + +This transport connects directly to a FastMCP server instance in the same +Python process. It works with both FastMCP 2.x servers and FastMCP 1.0 +servers from the low-level MCP SDK. This is particularly useful for unit +tests or scenarios where client and server run in the same runtime. + + +**Methods:** + +#### `connect_session` + +```python +connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] +``` diff --git a/docs/python-sdk/fastmcp-client-transports-stdio.mdx b/docs/python-sdk/fastmcp-client-transports-stdio.mdx new file mode 100644 index 000000000..fc815de7b --- /dev/null +++ b/docs/python-sdk/fastmcp-client-transports-stdio.mdx @@ -0,0 +1,79 @@ +--- +title: stdio +sidebarTitle: stdio +--- + +# `fastmcp.client.transports.stdio` + +## Classes + +### `StdioTransport` + + +Base transport for connecting to an MCP server via subprocess with stdio. + +This is a base class that can be subclassed for specific command-based +transports like Python, Node, Uvx, etc. + + +**Methods:** + +#### `connect_session` + +```python +connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] +``` + +#### `connect` + +```python +connect(self, **session_kwargs: Unpack[SessionKwargs]) -> ClientSession | None +``` + +#### `disconnect` + +```python +disconnect(self) +``` + +#### `close` + +```python +close(self) +``` + +### `PythonStdioTransport` + + +Transport for running Python scripts. + + +### `FastMCPStdioTransport` + + +Transport for running FastMCP servers using the FastMCP CLI. + + +### `NodeStdioTransport` + + +Transport for running Node.js scripts. + + +### `UvStdioTransport` + + +Transport for running commands via the uv tool. + + +### `UvxStdioTransport` + + +Transport for running commands via the uvx tool. + + +### `NpxStdioTransport` + + +Transport for running commands via the npx tool. + diff --git a/docs/python-sdk/fastmcp-client-transports.mdx b/docs/python-sdk/fastmcp-client-transports.mdx deleted file mode 100644 index 1d84977af..000000000 --- a/docs/python-sdk/fastmcp-client-transports.mdx +++ /dev/null @@ -1,306 +0,0 @@ ---- -title: transports -sidebarTitle: transports ---- - -# `fastmcp.client.transports` - -## Functions - -### `infer_transport` - -```python -infer_transport(transport: ClientTransport | FastMCP | FastMCP1Server | AnyUrl | Path | MCPConfig | dict[str, Any] | str) -> ClientTransport -``` - - -Infer the appropriate transport type from the given transport argument. - -This function attempts to infer the correct transport type from the provided -argument, handling various input types and converting them to the appropriate -ClientTransport subclass. - -The function supports these input types: -- ClientTransport: Used directly without modification -- FastMCP or FastMCP1Server: Creates an in-memory FastMCPTransport -- Path or str (file path): Creates PythonStdioTransport (.py) or NodeStdioTransport (.js) -- AnyUrl or str (URL): Creates StreamableHttpTransport (default) or SSETransport (for /sse endpoints) -- MCPConfig or dict: Creates MCPConfigTransport, potentially connecting to multiple servers - -For HTTP URLs, they are assumed to be Streamable HTTP URLs unless they end in `/sse`. - -For MCPConfig with multiple servers, a composite client is created where each server -is mounted with its name as prefix. This allows accessing tools and resources from multiple -servers through a single unified client interface, using naming patterns like -`servername_toolname` for tools and `protocol://servername/path` for resources. -If the MCPConfig contains only one server, a direct connection is established without prefixing. - -**Examples:** - -```python -# Connect to a local Python script -transport = infer_transport("my_script.py") - -# Connect to a remote server via HTTP -transport = infer_transport("http://example.com/mcp") - -# Connect to multiple servers using MCPConfig -config = { - "mcpServers": { - "weather": {"url": "http://weather.example.com/mcp"}, - "calendar": {"url": "http://calendar.example.com/mcp"} - } -} -transport = infer_transport(config) -``` - - -## Classes - -### `SessionKwargs` - - -Keyword arguments for the MCP ClientSession constructor. - - -### `ClientTransport` - - -Abstract base class for different MCP client transport mechanisms. - -A Transport is responsible for establishing and managing connections -to an MCP server, and providing a ClientSession within an async context. - - -**Methods:** - -#### `connect_session` - -```python -connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] -``` - -Establishes a connection and yields an active ClientSession. - -The ClientSession is *not* expected to be initialized in this context manager. - -The session is guaranteed to be valid only within the scope of the -async context manager. Connection setup and teardown are handled -within this context. - -**Args:** -- `**session_kwargs`: Keyword arguments to pass to the ClientSession - constructor (e.g., callbacks, timeouts). - - -#### `close` - -```python -close(self) -``` - -Close the transport. - - -#### `get_session_id` - -```python -get_session_id(self) -> str | None -``` - -Get the session ID for this transport, if available. - - -### `SSETransport` - - -Transport implementation that connects to an MCP server via Server-Sent Events. - - -**Methods:** - -#### `connect_session` - -```python -connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] -``` - -### `StreamableHttpTransport` - - -Transport implementation that connects to an MCP server via Streamable HTTP Requests. - - -**Methods:** - -#### `connect_session` - -```python -connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] -``` - -#### `get_session_id` - -```python -get_session_id(self) -> str | None -``` - -#### `close` - -```python -close(self) -``` - -### `StdioTransport` - - -Base transport for connecting to an MCP server via subprocess with stdio. - -This is a base class that can be subclassed for specific command-based -transports like Python, Node, Uvx, etc. - - -**Methods:** - -#### `connect_session` - -```python -connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] -``` - -#### `connect` - -```python -connect(self, **session_kwargs: Unpack[SessionKwargs]) -> ClientSession | None -``` - -#### `disconnect` - -```python -disconnect(self) -``` - -#### `close` - -```python -close(self) -``` - -### `PythonStdioTransport` - - -Transport for running Python scripts. - - -### `FastMCPStdioTransport` - - -Transport for running FastMCP servers using the FastMCP CLI. - - -### `NodeStdioTransport` - - -Transport for running Node.js scripts. - - -### `UvStdioTransport` - - -Transport for running commands via the uv tool. - - -### `UvxStdioTransport` - - -Transport for running commands via the uvx tool. - - -### `NpxStdioTransport` - - -Transport for running commands via the npx tool. - - -### `FastMCPTransport` - - -In-memory transport for FastMCP servers. - -This transport connects directly to a FastMCP server instance in the same -Python process. It works with both FastMCP 2.x servers and FastMCP 1.0 -servers from the low-level MCP SDK. This is particularly useful for unit -tests or scenarios where client and server run in the same runtime. - - -**Methods:** - -#### `connect_session` - -```python -connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] -``` - -### `MCPConfigTransport` - - -Transport for connecting to one or more MCP servers defined in an MCPConfig. - -This transport provides a unified interface to multiple MCP servers defined in an MCPConfig -object or dictionary matching the MCPConfig schema. It supports two key scenarios: - -1. If the MCPConfig contains exactly one server, it creates a direct transport to that server. -2. If the MCPConfig contains multiple servers, it creates a composite client by mounting - all servers on a single FastMCP instance, with each server's name, by default, used as its mounting prefix. - -In the multi-server case, tools are accessible with the prefix pattern `{server_name}_{tool_name}` -and resources with the pattern `protocol://{server_name}/path/to/resource`. - -This is particularly useful for creating clients that need to interact with multiple specialized -MCP servers through a single interface, simplifying client code. - -**Examples:** - -```python -from fastmcp import Client - -# Create a config with multiple servers -config = { - "mcpServers": { - "weather": { - "url": "https://weather-api.example.com/mcp", - "transport": "http" - }, - "calendar": { - "url": "https://calendar-api.example.com/mcp", - "transport": "http" - } - } -} - -# Create a client with the config -client = Client(config) - -async with client: - # Access tools with prefixes - weather = await client.call_tool("weather_get_forecast", {"city": "London"}) - events = await client.call_tool("calendar_list_events", {"date": "2023-06-01"}) - - # Access resources with prefixed URIs - icons = await client.read_resource("weather://weather/icons/sunny") -``` - - -**Methods:** - -#### `connect_session` - -```python -connect_session(self, **session_kwargs: Unpack[SessionKwargs]) -> AsyncIterator[ClientSession] -``` - -#### `close` - -```python -close(self) -``` diff --git a/justfile b/justfile index 6fd10bf58..28f56c2e0 100644 --- a/justfile +++ b/justfile @@ -16,7 +16,7 @@ docs: # Check for broken links in documentation docs-broken-links: - cd docs && npx --yes mint@latest install && npx --yes mint@latest broken-links + cd docs && npx --yes mint@latest broken-links # Generate API reference documentation for all modules api-ref-all: diff --git a/loq.toml b/loq.toml index 52222dfd2..9726aba25 100644 --- a/loq.toml +++ b/loq.toml @@ -78,10 +78,6 @@ max_lines = 1584 path = "src/fastmcp/server/auth/oauth_proxy.py" max_lines = 2282 -[[rules]] -path = "src/fastmcp/client/transports.py" -max_lines = 1208 - [[rules]] path = "tests/server/test_dependencies.py" max_lines = 1046 diff --git a/src/fastmcp/client/transports.py b/src/fastmcp/client/transports.py deleted file mode 100644 index 50b62106c..000000000 --- a/src/fastmcp/client/transports.py +++ /dev/null @@ -1,1211 +0,0 @@ -import abc -import asyncio -import contextlib -import datetime -import os -import shutil -import sys -import warnings -from collections.abc import AsyncIterator, Callable -from pathlib import Path -from typing import Any, Literal, TextIO, TypeVar, cast, overload - -import anyio -import httpx -import mcp.types -from mcp import ClientSession, StdioServerParameters -from mcp.client.session import ( - ElicitationFnT, - ListRootsFnT, - LoggingFnT, - MessageHandlerFnT, - SamplingFnT, -) -from mcp.client.sse import sse_client -from mcp.client.stdio import stdio_client -from mcp.client.streamable_http import streamable_http_client -from mcp.server.fastmcp import FastMCP as FastMCP1Server -from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client -from mcp.shared.memory import create_client_server_memory_streams -from pydantic import AnyUrl -from typing_extensions import TypedDict, Unpack - -import fastmcp -from fastmcp.client.auth.bearer import BearerAuth -from fastmcp.client.auth.oauth import OAuth -from fastmcp.mcp_config import ( - MCPConfig, - MCPServerTypes, - RemoteMCPServer, - StdioMCPServer, - TransformingRemoteMCPServer, - TransformingStdioMCPServer, - infer_transport_type_from_url, -) -from fastmcp.server.dependencies import get_http_headers -from fastmcp.server.server import FastMCP, create_proxy -from fastmcp.utilities.logging import get_logger -from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment - -logger = get_logger(__name__) - -# TypeVar for preserving specific ClientTransport subclass types -ClientTransportT = TypeVar("ClientTransportT", bound="ClientTransport") - -__all__ = [ - "ClientTransport", - "FastMCPStdioTransport", - "FastMCPTransport", - "NodeStdioTransport", - "NpxStdioTransport", - "PythonStdioTransport", - "SSETransport", - "StdioTransport", - "StreamableHttpTransport", - "UvStdioTransport", - "UvxStdioTransport", - "infer_transport", -] - - -class SessionKwargs(TypedDict, total=False): - """Keyword arguments for the MCP ClientSession constructor.""" - - read_timeout_seconds: datetime.timedelta | None - sampling_callback: SamplingFnT | None - sampling_capabilities: mcp.types.SamplingCapability | None - list_roots_callback: ListRootsFnT | None - logging_callback: LoggingFnT | None - elicitation_callback: ElicitationFnT | None - message_handler: MessageHandlerFnT | None - client_info: mcp.types.Implementation | None - - -class ClientTransport(abc.ABC): - """ - Abstract base class for different MCP client transport mechanisms. - - A Transport is responsible for establishing and managing connections - to an MCP server, and providing a ClientSession within an async context. - - """ - - @abc.abstractmethod - @contextlib.asynccontextmanager - async def connect_session( - self, **session_kwargs: Unpack[SessionKwargs] - ) -> AsyncIterator[ClientSession]: - """ - Establishes a connection and yields an active ClientSession. - - The ClientSession is *not* expected to be initialized in this context manager. - - The session is guaranteed to be valid only within the scope of the - async context manager. Connection setup and teardown are handled - within this context. - - Args: - **session_kwargs: Keyword arguments to pass to the ClientSession - constructor (e.g., callbacks, timeouts). - - Yields: - A mcp.ClientSession instance. - """ - raise NotImplementedError - yield - - def __repr__(self) -> str: - # Basic representation for subclasses - return f"<{self.__class__.__name__}>" - - async def close(self): # noqa: B027 - """Close the transport.""" - - def get_session_id(self) -> str | None: - """Get the session ID for this transport, if available.""" - return None - - def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None): - if auth is not None: - raise ValueError("This transport does not support auth") - - -class SSETransport(ClientTransport): - """Transport implementation that connects to an MCP server via Server-Sent Events.""" - - def __init__( - self, - url: str | AnyUrl, - headers: dict[str, str] | None = None, - auth: httpx.Auth | Literal["oauth"] | str | None = None, - sse_read_timeout: datetime.timedelta | float | int | None = None, - httpx_client_factory: McpHttpClientFactory | None = None, - ): - if isinstance(url, AnyUrl): - url = str(url) - if not isinstance(url, str) or not url.startswith("http"): - raise ValueError("Invalid HTTP/S URL provided for SSE.") - - # Don't modify the URL path - respect the exact URL provided by the user - # Some servers are strict about trailing slashes (e.g., PayPal MCP) - - self.url = url - self.headers = headers or {} - self.httpx_client_factory = httpx_client_factory - self._set_auth(auth) - - if isinstance(sse_read_timeout, int | float): - sse_read_timeout = datetime.timedelta(seconds=float(sse_read_timeout)) - self.sse_read_timeout = sse_read_timeout - - def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None): - if auth == "oauth": - auth = OAuth(self.url, httpx_client_factory=self.httpx_client_factory) - elif isinstance(auth, str): - auth = BearerAuth(auth) - self.auth = auth - - @contextlib.asynccontextmanager - async def connect_session( - self, **session_kwargs: Unpack[SessionKwargs] - ) -> AsyncIterator[ClientSession]: - client_kwargs: dict[str, Any] = {} - - # load headers from an active HTTP request, if available. This will only be true - # if the client is used in a FastMCP Proxy, in which case the MCP client headers - # need to be forwarded to the remote server. - client_kwargs["headers"] = get_http_headers() | self.headers - - # sse_read_timeout has a default value set, so we can't pass None without overriding it - # instead we simply leave the kwarg out if it's not provided - if self.sse_read_timeout is not None: - client_kwargs["sse_read_timeout"] = self.sse_read_timeout.total_seconds() - if session_kwargs.get("read_timeout_seconds") is not None: - read_timeout_seconds = cast( - datetime.timedelta, session_kwargs.get("read_timeout_seconds") - ) - client_kwargs["timeout"] = read_timeout_seconds.total_seconds() - - if self.httpx_client_factory is not None: - client_kwargs["httpx_client_factory"] = self.httpx_client_factory - - async with sse_client(self.url, auth=self.auth, **client_kwargs) as transport: - read_stream, write_stream = transport - async with ClientSession( - read_stream, write_stream, **session_kwargs - ) as session: - yield session - - def __repr__(self) -> str: - return f"" - - -class StreamableHttpTransport(ClientTransport): - """Transport implementation that connects to an MCP server via Streamable HTTP Requests.""" - - def __init__( - self, - url: str | AnyUrl, - headers: dict[str, str] | None = None, - auth: httpx.Auth | Literal["oauth"] | str | None = None, - sse_read_timeout: datetime.timedelta | float | int | None = None, - httpx_client_factory: McpHttpClientFactory | None = None, - ): - """Initialize a Streamable HTTP transport. - - Args: - url: The MCP server endpoint URL. - headers: Optional headers to include in requests. - auth: Authentication method - httpx.Auth, "oauth" for OAuth flow, - or a bearer token string. - sse_read_timeout: Deprecated. Use read_timeout_seconds in session_kwargs. - httpx_client_factory: Optional factory for creating httpx.AsyncClient. - If provided, must accept keyword arguments: headers, auth, - follow_redirects, and optionally timeout. Using **kwargs is - recommended to ensure forward compatibility. - """ - if isinstance(url, AnyUrl): - url = str(url) - if not isinstance(url, str) or not url.startswith("http"): - raise ValueError("Invalid HTTP/S URL provided for Streamable HTTP.") - - # Don't modify the URL path - respect the exact URL provided by the user - # Some servers are strict about trailing slashes (e.g., PayPal MCP) - - self.url = url - self.headers = headers or {} - self.httpx_client_factory = httpx_client_factory - self._set_auth(auth) - - if sse_read_timeout is not None: - if fastmcp.settings.deprecation_warnings: - warnings.warn( - "The `sse_read_timeout` parameter is deprecated and no longer used. " - "The new streamable_http_client API does not support this parameter. " - "Use `read_timeout_seconds` in session_kwargs or configure timeout on " - "the httpx client via `httpx_client_factory` instead.", - DeprecationWarning, - stacklevel=2, - ) - if isinstance(sse_read_timeout, int | float): - sse_read_timeout = datetime.timedelta(seconds=float(sse_read_timeout)) - self.sse_read_timeout = sse_read_timeout - - self._get_session_id_cb: Callable[[], str | None] | None = None - - def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None): - if auth == "oauth": - auth = OAuth(self.url, httpx_client_factory=self.httpx_client_factory) - elif isinstance(auth, str): - auth = BearerAuth(auth) - self.auth = auth - - @contextlib.asynccontextmanager - async def connect_session( - self, **session_kwargs: Unpack[SessionKwargs] - ) -> AsyncIterator[ClientSession]: - # Load headers from an active HTTP request, if available. This will only be true - # if the client is used in a FastMCP Proxy, in which case the MCP client headers - # need to be forwarded to the remote server. - headers = get_http_headers() | self.headers - - # Configure timeout if provided, preserving MCP's 30s connect default - timeout: httpx.Timeout | None = None - if session_kwargs.get("read_timeout_seconds") is not None: - read_timeout_seconds = cast( - datetime.timedelta, session_kwargs.get("read_timeout_seconds") - ) - timeout = httpx.Timeout(30.0, read=read_timeout_seconds.total_seconds()) - - # Create httpx client from factory or use default with MCP-appropriate timeouts - # create_mcp_http_client uses 30s connect/5min read timeout by default, - # and always enables follow_redirects - if self.httpx_client_factory is not None: - # Factory clients get the full kwargs for backwards compatibility - http_client = self.httpx_client_factory( - headers=headers, - auth=self.auth, - follow_redirects=True, # type: ignore[call-arg] - **({"timeout": timeout} if timeout else {}), - ) - else: - http_client = create_mcp_http_client( - headers=headers, - timeout=timeout, - auth=self.auth, - ) - - # Ensure httpx client is closed after use - async with ( - http_client, - streamable_http_client(self.url, http_client=http_client) as transport, - ): - read_stream, write_stream, get_session_id = transport - self._get_session_id_cb = get_session_id - async with ClientSession( - read_stream, write_stream, **session_kwargs - ) as session: - yield session - - def get_session_id(self) -> str | None: - if self._get_session_id_cb: - try: - return self._get_session_id_cb() - except Exception: - return None - return None - - async def close(self): - # Reset the session id callback - self._get_session_id_cb = None - - def __repr__(self) -> str: - return f"" - - -class StdioTransport(ClientTransport): - """ - Base transport for connecting to an MCP server via subprocess with stdio. - - This is a base class that can be subclassed for specific command-based - transports like Python, Node, Uvx, etc. - """ - - def __init__( - self, - command: str, - args: list[str], - env: dict[str, str] | None = None, - cwd: str | None = None, - keep_alive: bool | None = None, - log_file: Path | TextIO | None = None, - ): - """ - Initialize a Stdio transport. - - Args: - command: The command to run (e.g., "python", "node", "uvx") - args: The arguments to pass to the command - env: Environment variables to set for the subprocess - cwd: Current working directory for the subprocess - keep_alive: Whether to keep the subprocess alive between connections. - Defaults to True. When True, the subprocess remains active - after the connection context exits, allowing reuse in - subsequent connections. - log_file: Optional path or file-like object where subprocess stderr will - be written. Can be a Path or TextIO object. Defaults to sys.stderr - if not provided. When a Path is provided, the file will be created - if it doesn't exist, or appended to if it does. When set, server - errors will be written to this file instead of appearing in the console. - """ - self.command = command - self.args = args - self.env = env - self.cwd = cwd - if keep_alive is None: - keep_alive = True - self.keep_alive = keep_alive - self.log_file = log_file - - self._session: ClientSession | None = None - self._connect_task: asyncio.Task | None = None - self._ready_event = anyio.Event() - self._stop_event = anyio.Event() - - @contextlib.asynccontextmanager - async def connect_session( - self, **session_kwargs: Unpack[SessionKwargs] - ) -> AsyncIterator[ClientSession]: - try: - await self.connect(**session_kwargs) - yield cast(ClientSession, self._session) - finally: - if not self.keep_alive: - await self.disconnect() - else: - logger.debug("Stdio transport has keep_alive=True, not disconnecting") - - async def connect( - self, **session_kwargs: Unpack[SessionKwargs] - ) -> ClientSession | None: - if self._connect_task is not None: - return - - session_future: asyncio.Future[ClientSession] = asyncio.Future() - - # start the connection task - self._connect_task = asyncio.create_task( - _stdio_transport_connect_task( - command=self.command, - args=self.args, - env=self.env, - cwd=self.cwd, - log_file=self.log_file, - # TODO(ty): remove when ty supports Unpack[TypedDict] inference - session_kwargs=session_kwargs, # type: ignore[arg-type] - ready_event=self._ready_event, - stop_event=self._stop_event, - session_future=session_future, - ) - ) - - # wait for the client to be ready before returning - await self._ready_event.wait() - - # Check if connect task completed with an exception (early failure) - if self._connect_task.done(): - exception = self._connect_task.exception() - if exception is not None: - raise exception - - self._session = await session_future - return self._session - - async def disconnect(self): - if self._connect_task is None: - return - - # signal the connection task to stop - self._stop_event.set() - - # wait for the connection task to finish cleanly - await self._connect_task - - # reset variables and events for potential future reconnects - self._connect_task = None - self._stop_event = anyio.Event() - self._ready_event = anyio.Event() - - async def close(self): - await self.disconnect() - - def __del__(self): - """Ensure that we send a disconnection signal to the transport task if we are being garbage collected.""" - if not self._stop_event.is_set(): - self._stop_event.set() - - def __repr__(self) -> str: - return ( - f"<{self.__class__.__name__}(command='{self.command}', args={self.args})>" - ) - - -async def _stdio_transport_connect_task( - command: str, - args: list[str], - env: dict[str, str] | None, - cwd: str | None, - log_file: Path | TextIO | None, - session_kwargs: SessionKwargs, - ready_event: anyio.Event, - stop_event: anyio.Event, - session_future: asyncio.Future[ClientSession], -): - """A standalone connection task for a stdio transport. It is not a part of the StdioTransport class - to ensure that the connection task does not hold a reference to the Transport object.""" - - try: - async with contextlib.AsyncExitStack() as stack: - try: - server_params = StdioServerParameters( - command=command, - args=args, - env=env, - cwd=cwd, - ) - # Handle log_file: Path needs to be opened, TextIO used as-is - if log_file is None: - log_file_handle = sys.stderr - elif isinstance(log_file, Path): - log_file_handle = stack.enter_context(log_file.open("a")) - else: - # Must be TextIO - use it directly - log_file_handle = log_file - - transport = await stack.enter_async_context( - stdio_client(server_params, errlog=log_file_handle) - ) - read_stream, write_stream = transport - session_future.set_result( - await stack.enter_async_context( - ClientSession(read_stream, write_stream, **session_kwargs) - ) - ) - - logger.debug("Stdio transport connected") - ready_event.set() - - # Wait until disconnect is requested (stop_event is set) - await stop_event.wait() - finally: - # Clean up client on exit - logger.debug("Stdio transport disconnected") - except Exception: - # Ensure ready event is set even if connection fails - ready_event.set() - raise - - -class PythonStdioTransport(StdioTransport): - """Transport for running Python scripts.""" - - def __init__( - self, - script_path: str | Path, - args: list[str] | None = None, - env: dict[str, str] | None = None, - cwd: str | None = None, - python_cmd: str = sys.executable, - keep_alive: bool | None = None, - log_file: Path | TextIO | None = None, - ): - """ - Initialize a Python transport. - - Args: - script_path: Path to the Python script to run - args: Additional arguments to pass to the script - env: Environment variables to set for the subprocess - cwd: Current working directory for the subprocess - python_cmd: Python command to use (default: "python") - keep_alive: Whether to keep the subprocess alive between connections. - Defaults to True. When True, the subprocess remains active - after the connection context exits, allowing reuse in - subsequent connections. - log_file: Optional path or file-like object where subprocess stderr will - be written. Can be a Path or TextIO object. Defaults to sys.stderr - if not provided. When a Path is provided, the file will be created - if it doesn't exist, or appended to if it does. When set, server - errors will be written to this file instead of appearing in the console. - """ - script_path = Path(script_path).resolve() - if not script_path.is_file(): - raise FileNotFoundError(f"Script not found: {script_path}") - if not str(script_path).endswith(".py"): - raise ValueError(f"Not a Python script: {script_path}") - - full_args = [str(script_path)] - if args: - full_args.extend(args) - - super().__init__( - command=python_cmd, - args=full_args, - env=env, - cwd=cwd, - keep_alive=keep_alive, - log_file=log_file, - ) - self.script_path = script_path - - -class FastMCPStdioTransport(StdioTransport): - """Transport for running FastMCP servers using the FastMCP CLI.""" - - def __init__( - self, - script_path: str | Path, - args: list[str] | None = None, - env: dict[str, str] | None = None, - cwd: str | None = None, - keep_alive: bool | None = None, - log_file: Path | TextIO | None = None, - ): - script_path = Path(script_path).resolve() - if not script_path.is_file(): - raise FileNotFoundError(f"Script not found: {script_path}") - if not str(script_path).endswith(".py"): - raise ValueError(f"Not a Python script: {script_path}") - - super().__init__( - command="fastmcp", - args=["run", str(script_path)], - env=env, - cwd=cwd, - keep_alive=keep_alive, - log_file=log_file, - ) - self.script_path = script_path - - -class NodeStdioTransport(StdioTransport): - """Transport for running Node.js scripts.""" - - def __init__( - self, - script_path: str | Path, - args: list[str] | None = None, - env: dict[str, str] | None = None, - cwd: str | None = None, - node_cmd: str = "node", - keep_alive: bool | None = None, - log_file: Path | TextIO | None = None, - ): - """ - Initialize a Node transport. - - Args: - script_path: Path to the Node.js script to run - args: Additional arguments to pass to the script - env: Environment variables to set for the subprocess - cwd: Current working directory for the subprocess - node_cmd: Node.js command to use (default: "node") - keep_alive: Whether to keep the subprocess alive between connections. - Defaults to True. When True, the subprocess remains active - after the connection context exits, allowing reuse in - subsequent connections. - log_file: Optional path or file-like object where subprocess stderr will - be written. Can be a Path or TextIO object. Defaults to sys.stderr - if not provided. When a Path is provided, the file will be created - if it doesn't exist, or appended to if it does. When set, server - errors will be written to this file instead of appearing in the console. - """ - script_path = Path(script_path).resolve() - if not script_path.is_file(): - raise FileNotFoundError(f"Script not found: {script_path}") - if not str(script_path).endswith(".js"): - raise ValueError(f"Not a JavaScript script: {script_path}") - - full_args = [str(script_path)] - if args: - full_args.extend(args) - - super().__init__( - command=node_cmd, - args=full_args, - env=env, - cwd=cwd, - keep_alive=keep_alive, - log_file=log_file, - ) - self.script_path = script_path - - -class UvStdioTransport(StdioTransport): - """Transport for running commands via the uv tool.""" - - def __init__( - self, - command: str, - args: list[str] | None = None, - module: bool = False, - project_directory: Path | None = None, - python_version: str | None = None, - with_packages: list[str] | None = None, - with_requirements: Path | None = None, - env_vars: dict[str, str] | None = None, - keep_alive: bool | None = None, - ): - # Basic validation - if project_directory and not project_directory.exists(): - raise NotADirectoryError( - f"Project directory not found: {project_directory}" - ) - - # Create Environment from provided parameters (internal use) - env_config = UVEnvironment( - python=python_version, - dependencies=with_packages, - requirements=with_requirements, - project=project_directory, - editable=None, # Not exposed in this transport - ) - - # Build uv arguments using the config - uv_args: list[str] = [] - - # Check if we need any environment setup - if env_config._must_run_with_uv(): - # Use the config to build args, but we need to handle the command differently - # since transport has specific needs - uv_args = ["run"] - - if python_version: - uv_args.extend(["--python", python_version]) - if project_directory: - uv_args.extend(["--directory", str(project_directory)]) - - # Note: Don't add fastmcp as dependency here, transport is for general use - for pkg in with_packages or []: - uv_args.extend(["--with", pkg]) - if with_requirements: - uv_args.extend(["--with-requirements", str(with_requirements)]) - else: - # No environment setup needed - uv_args = ["run"] - - if module: - uv_args.append("--module") - - if not args: - args = [] - - uv_args.extend([command, *args]) - - # Get environment with any additional variables - env: dict[str, str] | None = None - if env_vars or project_directory: - env = os.environ.copy() - if project_directory: - env["UV_PROJECT_DIR"] = str(project_directory) - if env_vars: - env.update(env_vars) - - super().__init__( - command="uv", - args=uv_args, - env=env, - cwd=None, # Use --directory flag instead of cwd - keep_alive=keep_alive, - ) - - -class UvxStdioTransport(StdioTransport): - """Transport for running commands via the uvx tool.""" - - def __init__( - self, - tool_name: str, - tool_args: list[str] | None = None, - project_directory: str | None = None, - python_version: str | None = None, - with_packages: list[str] | None = None, - from_package: str | None = None, - env_vars: dict[str, str] | None = None, - keep_alive: bool | None = None, - ): - """ - Initialize a Uvx transport. - - Args: - tool_name: Name of the tool to run via uvx - tool_args: Arguments to pass to the tool - project_directory: Project directory (for package resolution) - python_version: Python version to use - with_packages: Additional packages to include - from_package: Package to install the tool from - env_vars: Additional environment variables - keep_alive: Whether to keep the subprocess alive between connections. - Defaults to True. When True, the subprocess remains active - after the connection context exits, allowing reuse in - subsequent connections. - """ - # Basic validation - if project_directory and not Path(project_directory).exists(): - raise NotADirectoryError( - f"Project directory not found: {project_directory}" - ) - - # Build uvx arguments - uvx_args: list[str] = [] - if python_version: - uvx_args.extend(["--python", python_version]) - if from_package: - uvx_args.extend(["--from", from_package]) - for pkg in with_packages or []: - uvx_args.extend(["--with", pkg]) - - # Add the tool name and tool args - uvx_args.append(tool_name) - if tool_args: - uvx_args.extend(tool_args) - - env: dict[str, str] | None = None - if env_vars: - env = os.environ.copy() - env.update(env_vars) - - super().__init__( - command="uvx", - args=uvx_args, - env=env, - cwd=project_directory, - keep_alive=keep_alive, - ) - self.tool_name: str = tool_name - - -class NpxStdioTransport(StdioTransport): - """Transport for running commands via the npx tool.""" - - def __init__( - self, - package: str, - args: list[str] | None = None, - project_directory: str | None = None, - env_vars: dict[str, str] | None = None, - use_package_lock: bool = True, - keep_alive: bool | None = None, - ): - """ - Initialize an Npx transport. - - Args: - package: Name of the npm package to run - args: Arguments to pass to the package command - project_directory: Project directory with package.json - env_vars: Additional environment variables - use_package_lock: Whether to use package-lock.json (--prefer-offline) - keep_alive: Whether to keep the subprocess alive between connections. - Defaults to True. When True, the subprocess remains active - after the connection context exits, allowing reuse in - subsequent connections. - """ - # verify npx is installed - if shutil.which("npx") is None: - raise ValueError("Command 'npx' not found") - - # Basic validation - if project_directory and not Path(project_directory).exists(): - raise NotADirectoryError( - f"Project directory not found: {project_directory}" - ) - - # Build npx arguments - npx_args = [] - if use_package_lock: - npx_args.append("--prefer-offline") - - # Add the package name and args - npx_args.append(package) - if args: - npx_args.extend(args) - - # Get environment with any additional variables - env = None - if env_vars: - env = os.environ.copy() - env.update(env_vars) - - super().__init__( - command="npx", - args=npx_args, - env=env, - cwd=project_directory, - keep_alive=keep_alive, - ) - self.package = package - - -class FastMCPTransport(ClientTransport): - """In-memory transport for FastMCP servers. - - This transport connects directly to a FastMCP server instance in the same - Python process. It works with both FastMCP 2.x servers and FastMCP 1.0 - servers from the low-level MCP SDK. This is particularly useful for unit - tests or scenarios where client and server run in the same runtime. - """ - - def __init__(self, mcp: FastMCP | FastMCP1Server, raise_exceptions: bool = False): - """Initialize a FastMCPTransport from a FastMCP server instance.""" - - # Accept both FastMCP 2.x and FastMCP 1.0 servers. Both expose a - # ``_mcp_server`` attribute pointing to the underlying MCP server - # implementation, so we can treat them identically. - self.server = mcp - self.raise_exceptions = raise_exceptions - - @contextlib.asynccontextmanager - async def connect_session( - self, **session_kwargs: Unpack[SessionKwargs] - ) -> AsyncIterator[ClientSession]: - async with create_client_server_memory_streams() as ( - client_streams, - server_streams, - ): - client_read, client_write = client_streams - server_read, server_write = server_streams - - # Capture exceptions to re-raise after task group cleanup. - # anyio task groups can suppress exceptions when cancel_scope.cancel() - # is called during cleanup, so we capture and re-raise manually. - exception_to_raise: BaseException | None = None - - async with ( - anyio.create_task_group() as tg, - _enter_server_lifespan(server=self.server), - ): - tg.start_soon( - lambda: self.server._mcp_server.run( - server_read, - server_write, - self.server._mcp_server.create_initialization_options(), - raise_exceptions=self.raise_exceptions, - ) - ) - - try: - async with ClientSession( - read_stream=client_read, - write_stream=client_write, - **session_kwargs, - ) as client_session: - yield client_session - except BaseException as e: - exception_to_raise = e - finally: - tg.cancel_scope.cancel() - - # Re-raise after task group has exited cleanly - if exception_to_raise is not None: - raise exception_to_raise - - def __repr__(self) -> str: - return f"" - - -@contextlib.asynccontextmanager -async def _enter_server_lifespan( - server: FastMCP | FastMCP1Server, -) -> AsyncIterator[None]: - """Enters the server's lifespan context for FastMCP servers and does nothing for FastMCP 1 servers.""" - if isinstance(server, FastMCP): - async with server._lifespan_manager(): - yield - else: - yield - - -class MCPConfigTransport(ClientTransport): - """Transport for connecting to one or more MCP servers defined in an MCPConfig. - - This transport provides a unified interface to multiple MCP servers defined in an MCPConfig - object or dictionary matching the MCPConfig schema. It supports two key scenarios: - - 1. If the MCPConfig contains exactly one server, it creates a direct transport to that server. - 2. If the MCPConfig contains multiple servers, it creates a composite client by mounting - all servers on a single FastMCP instance, with each server's name, by default, used as its mounting prefix. - - In the multi-server case, tools are accessible with the prefix pattern `{server_name}_{tool_name}` - and resources with the pattern `protocol://{server_name}/path/to/resource`. - - This is particularly useful for creating clients that need to interact with multiple specialized - MCP servers through a single interface, simplifying client code. - - Examples: - ```python - from fastmcp import Client - - # Create a config with multiple servers - config = { - "mcpServers": { - "weather": { - "url": "https://weather-api.example.com/mcp", - "transport": "http" - }, - "calendar": { - "url": "https://calendar-api.example.com/mcp", - "transport": "http" - } - } - } - - # Create a client with the config - client = Client(config) - - async with client: - # Access tools with prefixes - weather = await client.call_tool("weather_get_forecast", {"city": "London"}) - events = await client.call_tool("calendar_list_events", {"date": "2023-06-01"}) - - # Access resources with prefixed URIs - icons = await client.read_resource("weather://weather/icons/sunny") - ``` - """ - - def __init__(self, config: MCPConfig | dict, name_as_prefix: bool = True): - if isinstance(config, dict): - config = MCPConfig.from_dict(config) - self.config = config - self.name_as_prefix = name_as_prefix - self._transports: list[ClientTransport] = [] - - if not self.config.mcpServers: - raise ValueError("No MCP servers defined in the config") - - # For single server, create transport eagerly so it can be inspected - if len(self.config.mcpServers) == 1: - self.transport = next(iter(self.config.mcpServers.values())).to_transport() - self._transports.append(self.transport) - - @contextlib.asynccontextmanager - async def connect_session( - self, **session_kwargs: Unpack[SessionKwargs] - ) -> AsyncIterator[ClientSession]: - # Single server - delegate directly to pre-created transport - if len(self.config.mcpServers) == 1: - async with self.transport.connect_session(**session_kwargs) as session: - yield session - return - - # Multiple servers - create composite with mounted proxies - # Close any previous transports from prior connections to avoid leaking - for t in self._transports: - await t.close() - self._transports = [] - timeout = session_kwargs.get("read_timeout_seconds") - composite = FastMCP[Any](name="MCPRouter") - - try: - for name, server_config in self.config.mcpServers.items(): - transport, proxy = self._create_proxy(name, server_config, timeout) - self._transports.append(transport) - composite.mount(proxy, namespace=name if self.name_as_prefix else None) - except Exception: - # Clean up any transports created before the failure - for t in self._transports: - await t.close() - self._transports = [] - raise - - async with FastMCPTransport(mcp=composite).connect_session( - **session_kwargs - ) as session: - yield session - - def _create_proxy( - self, - name: str, - config: MCPServerTypes, - timeout: datetime.timedelta | None, - ) -> tuple[ClientTransport, FastMCP[Any]]: - """Create underlying transport and proxy server for a single backend.""" - # Import here to avoid circular dependency - from fastmcp.server.providers.proxy import ProxyClient - - tool_transforms = None - include_tags = None - exclude_tags = None - - # Handle transforming servers - call base class to_transport() for underlying transport - if isinstance(config, TransformingStdioMCPServer): - transport = StdioMCPServer.to_transport(config) - tool_transforms = config.tools - include_tags = config.include_tags - exclude_tags = config.exclude_tags - elif isinstance(config, TransformingRemoteMCPServer): - transport = RemoteMCPServer.to_transport(config) - tool_transforms = config.tools - include_tags = config.include_tags - exclude_tags = config.exclude_tags - else: - transport = config.to_transport() - - client = ProxyClient(transport=transport, timeout=timeout) - # Create proxy without include_tags/exclude_tags - we'll add them after tool transforms - proxy = create_proxy( - client, - name=f"Proxy-{name}", - ) - # Add tool transforms FIRST - they may add/modify tags - if tool_transforms: - from fastmcp.server.transforms import ToolTransform - - proxy.add_transform(ToolTransform(tool_transforms)) - # Then add enabled filters - they filter based on tags - if include_tags: - proxy.enable(tags=set(include_tags), only=True) - if exclude_tags: - proxy.disable(tags=set(exclude_tags)) - return transport, proxy - - async def close(self): - for transport in self._transports: - await transport.close() - - def __repr__(self) -> str: - return f"" - - -@overload -def infer_transport(transport: ClientTransportT) -> ClientTransportT: ... - - -@overload -def infer_transport(transport: FastMCP) -> FastMCPTransport: ... - - -@overload -def infer_transport(transport: FastMCP1Server) -> FastMCPTransport: ... - - -@overload -def infer_transport(transport: MCPConfig) -> MCPConfigTransport: ... - - -@overload -def infer_transport(transport: dict[str, Any]) -> MCPConfigTransport: ... - - -@overload -def infer_transport( - transport: AnyUrl, -) -> SSETransport | StreamableHttpTransport: ... - - -@overload -def infer_transport( - transport: str, -) -> ( - PythonStdioTransport | NodeStdioTransport | SSETransport | StreamableHttpTransport -): ... - - -@overload -def infer_transport(transport: Path) -> PythonStdioTransport | NodeStdioTransport: ... - - -def infer_transport( - transport: ClientTransport - | FastMCP - | FastMCP1Server - | AnyUrl - | Path - | MCPConfig - | dict[str, Any] - | str, -) -> ClientTransport: - """ - Infer the appropriate transport type from the given transport argument. - - This function attempts to infer the correct transport type from the provided - argument, handling various input types and converting them to the appropriate - ClientTransport subclass. - - The function supports these input types: - - ClientTransport: Used directly without modification - - FastMCP or FastMCP1Server: Creates an in-memory FastMCPTransport - - Path or str (file path): Creates PythonStdioTransport (.py) or NodeStdioTransport (.js) - - AnyUrl or str (URL): Creates StreamableHttpTransport (default) or SSETransport (for /sse endpoints) - - MCPConfig or dict: Creates MCPConfigTransport, potentially connecting to multiple servers - - For HTTP URLs, they are assumed to be Streamable HTTP URLs unless they end in `/sse`. - - For MCPConfig with multiple servers, a composite client is created where each server - is mounted with its name as prefix. This allows accessing tools and resources from multiple - servers through a single unified client interface, using naming patterns like - `servername_toolname` for tools and `protocol://servername/path` for resources. - If the MCPConfig contains only one server, a direct connection is established without prefixing. - - Examples: - ```python - # Connect to a local Python script - transport = infer_transport("my_script.py") - - # Connect to a remote server via HTTP - transport = infer_transport("http://example.com/mcp") - - # Connect to multiple servers using MCPConfig - config = { - "mcpServers": { - "weather": {"url": "http://weather.example.com/mcp"}, - "calendar": {"url": "http://calendar.example.com/mcp"} - } - } - transport = infer_transport(config) - ``` - """ - - # the transport is already a ClientTransport - if isinstance(transport, ClientTransport): - return transport - - # the transport is a FastMCP server (2.x or 1.0) - elif isinstance(transport, FastMCP | FastMCP1Server): - inferred_transport = FastMCPTransport( - mcp=cast(FastMCP[Any] | FastMCP1Server, transport) - ) - - # the transport is a path to a script - elif isinstance(transport, Path | str) and Path(transport).exists(): - if str(transport).endswith(".py"): - inferred_transport = PythonStdioTransport(script_path=cast(Path, transport)) - elif str(transport).endswith(".js"): - inferred_transport = NodeStdioTransport(script_path=cast(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"): - inferred_transport_type = infer_transport_type_from_url( - cast(AnyUrl | str, transport) - ) - if inferred_transport_type == "sse": - inferred_transport = SSETransport(url=cast(AnyUrl | str, transport)) - else: - inferred_transport = StreamableHttpTransport( - url=cast(AnyUrl | str, transport) - ) - - # if the transport is a config dict or MCPConfig - elif isinstance(transport, dict | MCPConfig): - inferred_transport = MCPConfigTransport( - config=cast(dict | MCPConfig, 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/src/fastmcp/client/transports/__init__.py b/src/fastmcp/client/transports/__init__.py new file mode 100644 index 000000000..b3e9b7c38 --- /dev/null +++ b/src/fastmcp/client/transports/__init__.py @@ -0,0 +1,37 @@ +# Re-export all public APIs for backward compatibility +from mcp.server.fastmcp import FastMCP as FastMCP1Server + +from fastmcp.client.transports.base import ( + ClientTransport, + ClientTransportT, + SessionKwargs, +) +from fastmcp.client.transports.config import MCPConfigTransport +from fastmcp.client.transports.http import SSETransport, StreamableHttpTransport +from fastmcp.client.transports.inference import infer_transport +from fastmcp.client.transports.memory import FastMCPTransport +from fastmcp.client.transports.stdio import ( + FastMCPStdioTransport, + NodeStdioTransport, + NpxStdioTransport, + PythonStdioTransport, + StdioTransport, + UvStdioTransport, + UvxStdioTransport, +) +from fastmcp.server.server import FastMCP + +__all__ = [ + "ClientTransport", + "FastMCPStdioTransport", + "FastMCPTransport", + "NodeStdioTransport", + "NpxStdioTransport", + "PythonStdioTransport", + "SSETransport", + "StdioTransport", + "StreamableHttpTransport", + "UvStdioTransport", + "UvxStdioTransport", + "infer_transport", +] diff --git a/src/fastmcp/client/transports/base.py b/src/fastmcp/client/transports/base.py new file mode 100644 index 000000000..e9ebb1f3a --- /dev/null +++ b/src/fastmcp/client/transports/base.py @@ -0,0 +1,82 @@ +import abc +import contextlib +import datetime +from collections.abc import AsyncIterator +from typing import Literal, TypeVar + +import httpx +import mcp.types +from mcp import ClientSession +from mcp.client.session import ( + ElicitationFnT, + ListRootsFnT, + LoggingFnT, + MessageHandlerFnT, + SamplingFnT, +) +from typing_extensions import TypedDict, Unpack + +# TypeVar for preserving specific ClientTransport subclass types +ClientTransportT = TypeVar("ClientTransportT", bound="ClientTransport") + + +class SessionKwargs(TypedDict, total=False): + """Keyword arguments for the MCP ClientSession constructor.""" + + read_timeout_seconds: datetime.timedelta | None + sampling_callback: SamplingFnT | None + sampling_capabilities: mcp.types.SamplingCapability | None + list_roots_callback: ListRootsFnT | None + logging_callback: LoggingFnT | None + elicitation_callback: ElicitationFnT | None + message_handler: MessageHandlerFnT | None + client_info: mcp.types.Implementation | None + + +class ClientTransport(abc.ABC): + """ + Abstract base class for different MCP client transport mechanisms. + + A Transport is responsible for establishing and managing connections + to an MCP server, and providing a ClientSession within an async context. + + """ + + @abc.abstractmethod + @contextlib.asynccontextmanager + async def connect_session( + self, **session_kwargs: Unpack[SessionKwargs] + ) -> AsyncIterator[ClientSession]: + """ + Establishes a connection and yields an active ClientSession. + + The ClientSession is *not* expected to be initialized in this context manager. + + The session is guaranteed to be valid only within the scope of the + async context manager. Connection setup and teardown are handled + within this context. + + Args: + **session_kwargs: Keyword arguments to pass to the ClientSession + constructor (e.g., callbacks, timeouts). + + Yields: + A mcp.ClientSession instance. + """ + raise NotImplementedError + yield + + def __repr__(self) -> str: + # Basic representation for subclasses + return f"<{self.__class__.__name__}>" + + async def close(self): # noqa: B027 + """Close the transport.""" + + def get_session_id(self) -> str | None: + """Get the session ID for this transport, if available.""" + return None + + def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None): + if auth is not None: + raise ValueError("This transport does not support auth") diff --git a/src/fastmcp/client/transports/config.py b/src/fastmcp/client/transports/config.py new file mode 100644 index 000000000..084ef474d --- /dev/null +++ b/src/fastmcp/client/transports/config.py @@ -0,0 +1,170 @@ +import contextlib +import datetime +from collections.abc import AsyncIterator +from typing import Any + +from mcp import ClientSession +from typing_extensions import Unpack + +from fastmcp.client.transports.base import ClientTransport, SessionKwargs +from fastmcp.client.transports.memory import FastMCPTransport +from fastmcp.mcp_config import ( + MCPConfig, + MCPServerTypes, + RemoteMCPServer, + StdioMCPServer, + TransformingRemoteMCPServer, + TransformingStdioMCPServer, +) +from fastmcp.server.server import FastMCP, create_proxy + + +class MCPConfigTransport(ClientTransport): + """Transport for connecting to one or more MCP servers defined in an MCPConfig. + + This transport provides a unified interface to multiple MCP servers defined in an MCPConfig + object or dictionary matching the MCPConfig schema. It supports two key scenarios: + + 1. If the MCPConfig contains exactly one server, it creates a direct transport to that server. + 2. If the MCPConfig contains multiple servers, it creates a composite client by mounting + all servers on a single FastMCP instance, with each server's name, by default, used as its mounting prefix. + + In the multi-server case, tools are accessible with the prefix pattern `{server_name}_{tool_name}` + and resources with the pattern `protocol://{server_name}/path/to/resource`. + + This is particularly useful for creating clients that need to interact with multiple specialized + MCP servers through a single interface, simplifying client code. + + Examples: + ```python + from fastmcp import Client + + # Create a config with multiple servers + config = { + "mcpServers": { + "weather": { + "url": "https://weather-api.example.com/mcp", + "transport": "http" + }, + "calendar": { + "url": "https://calendar-api.example.com/mcp", + "transport": "http" + } + } + } + + # Create a client with the config + client = Client(config) + + async with client: + # Access tools with prefixes + weather = await client.call_tool("weather_get_forecast", {"city": "London"}) + events = await client.call_tool("calendar_list_events", {"date": "2023-06-01"}) + + # Access resources with prefixed URIs + icons = await client.read_resource("weather://weather/icons/sunny") + ``` + """ + + def __init__(self, config: MCPConfig | dict, name_as_prefix: bool = True): + if isinstance(config, dict): + config = MCPConfig.from_dict(config) + self.config = config + self.name_as_prefix = name_as_prefix + self._transports: list[ClientTransport] = [] + + if not self.config.mcpServers: + raise ValueError("No MCP servers defined in the config") + + # For single server, create transport eagerly so it can be inspected + if len(self.config.mcpServers) == 1: + self.transport = next(iter(self.config.mcpServers.values())).to_transport() + self._transports.append(self.transport) + + @contextlib.asynccontextmanager + async def connect_session( + self, **session_kwargs: Unpack[SessionKwargs] + ) -> AsyncIterator[ClientSession]: + # Single server - delegate directly to pre-created transport + if len(self.config.mcpServers) == 1: + async with self.transport.connect_session(**session_kwargs) as session: + yield session + return + + # Multiple servers - create composite with mounted proxies + # Close any previous transports from prior connections to avoid leaking + for t in self._transports: + await t.close() + self._transports = [] + timeout = session_kwargs.get("read_timeout_seconds") + composite = FastMCP[Any](name="MCPRouter") + + try: + for name, server_config in self.config.mcpServers.items(): + transport, proxy = self._create_proxy(name, server_config, timeout) + self._transports.append(transport) + composite.mount(proxy, namespace=name if self.name_as_prefix else None) + except Exception: + # Clean up any transports created before the failure + for t in self._transports: + await t.close() + self._transports = [] + raise + + async with FastMCPTransport(mcp=composite).connect_session( + **session_kwargs + ) as session: + yield session + + def _create_proxy( + self, + name: str, + config: MCPServerTypes, + timeout: datetime.timedelta | None, + ) -> tuple[ClientTransport, FastMCP[Any]]: + """Create underlying transport and proxy server for a single backend.""" + # Import here to avoid circular dependency + from fastmcp.server.providers.proxy import ProxyClient + + tool_transforms = None + include_tags = None + exclude_tags = None + + # Handle transforming servers - call base class to_transport() for underlying transport + if isinstance(config, TransformingStdioMCPServer): + transport = StdioMCPServer.to_transport(config) + tool_transforms = config.tools + include_tags = config.include_tags + exclude_tags = config.exclude_tags + elif isinstance(config, TransformingRemoteMCPServer): + transport = RemoteMCPServer.to_transport(config) + tool_transforms = config.tools + include_tags = config.include_tags + exclude_tags = config.exclude_tags + else: + transport = config.to_transport() + + client = ProxyClient(transport=transport, timeout=timeout) + # Create proxy without include_tags/exclude_tags - we'll add them after tool transforms + proxy = create_proxy( + client, + name=f"Proxy-{name}", + ) + # Add tool transforms FIRST - they may add/modify tags + if tool_transforms: + from fastmcp.server.transforms import ToolTransform + + proxy.add_transform(ToolTransform(tool_transforms)) + # Then add enabled filters - they filter based on tags + if include_tags: + proxy.enable(tags=set(include_tags), only=True) + if exclude_tags: + proxy.disable(tags=set(exclude_tags)) + return transport, proxy + + async def close(self): + for transport in self._transports: + await transport.close() + + def __repr__(self) -> str: + return f"" diff --git a/src/fastmcp/client/transports/http.py b/src/fastmcp/client/transports/http.py new file mode 100644 index 000000000..69baaeb46 --- /dev/null +++ b/src/fastmcp/client/transports/http.py @@ -0,0 +1,213 @@ +import contextlib +import datetime +from collections.abc import AsyncIterator, Callable +from typing import Any, Literal, cast + +import httpx +from mcp import ClientSession +from mcp.client.sse import sse_client +from mcp.client.streamable_http import streamable_http_client +from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client +from pydantic import AnyUrl +from typing_extensions import Unpack + +import fastmcp +from fastmcp.client.auth.bearer import BearerAuth +from fastmcp.client.auth.oauth import OAuth +from fastmcp.client.transports.base import ClientTransport, SessionKwargs +from fastmcp.server.dependencies import get_http_headers + + +class SSETransport(ClientTransport): + """Transport implementation that connects to an MCP server via Server-Sent Events.""" + + def __init__( + self, + url: str | AnyUrl, + headers: dict[str, str] | None = None, + auth: httpx.Auth | Literal["oauth"] | str | None = None, + sse_read_timeout: datetime.timedelta | float | int | None = None, + httpx_client_factory: McpHttpClientFactory | None = None, + ): + if isinstance(url, AnyUrl): + url = str(url) + if not isinstance(url, str) or not url.startswith("http"): + raise ValueError("Invalid HTTP/S URL provided for SSE.") + + # Don't modify the URL path - respect the exact URL provided by the user + # Some servers are strict about trailing slashes (e.g., PayPal MCP) + + self.url: str = url + self.headers = headers or {} + self.httpx_client_factory = httpx_client_factory + self._set_auth(auth) + + if isinstance(sse_read_timeout, int | float): + sse_read_timeout = datetime.timedelta(seconds=float(sse_read_timeout)) + self.sse_read_timeout = sse_read_timeout + + def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None): + if auth == "oauth": + auth = OAuth(self.url, httpx_client_factory=self.httpx_client_factory) + elif isinstance(auth, str): + auth = BearerAuth(auth) + self.auth = auth + + @contextlib.asynccontextmanager + async def connect_session( + self, **session_kwargs: Unpack[SessionKwargs] + ) -> AsyncIterator[ClientSession]: + client_kwargs: dict[str, Any] = {} + + # load headers from an active HTTP request, if available. This will only be true + # if the client is used in a FastMCP Proxy, in which case the MCP client headers + # need to be forwarded to the remote server. + client_kwargs["headers"] = get_http_headers() | self.headers + + # sse_read_timeout has a default value set, so we can't pass None without overriding it + # instead we simply leave the kwarg out if it's not provided + if self.sse_read_timeout is not None: + client_kwargs["sse_read_timeout"] = self.sse_read_timeout.total_seconds() + if session_kwargs.get("read_timeout_seconds") is not None: + read_timeout_seconds = cast( + datetime.timedelta, session_kwargs.get("read_timeout_seconds") + ) + client_kwargs["timeout"] = read_timeout_seconds.total_seconds() + + if self.httpx_client_factory is not None: + client_kwargs["httpx_client_factory"] = self.httpx_client_factory + + async with sse_client(self.url, auth=self.auth, **client_kwargs) as transport: + read_stream, write_stream = transport + async with ClientSession( + read_stream, write_stream, **session_kwargs + ) as session: + yield session + + def __repr__(self) -> str: + return f"" + + +class StreamableHttpTransport(ClientTransport): + """Transport implementation that connects to an MCP server via Streamable HTTP Requests.""" + + def __init__( + self, + url: str | AnyUrl, + headers: dict[str, str] | None = None, + auth: httpx.Auth | Literal["oauth"] | str | None = None, + sse_read_timeout: datetime.timedelta | float | int | None = None, + httpx_client_factory: McpHttpClientFactory | None = None, + ): + """Initialize a Streamable HTTP transport. + + Args: + url: The MCP server endpoint URL. + headers: Optional headers to include in requests. + auth: Authentication method - httpx.Auth, "oauth" for OAuth flow, + or a bearer token string. + sse_read_timeout: Deprecated. Use read_timeout_seconds in session_kwargs. + httpx_client_factory: Optional factory for creating httpx.AsyncClient. + If provided, must accept keyword arguments: headers, auth, + follow_redirects, and optionally timeout. Using **kwargs is + recommended to ensure forward compatibility. + """ + if isinstance(url, AnyUrl): + url = str(url) + if not isinstance(url, str) or not url.startswith("http"): + raise ValueError("Invalid HTTP/S URL provided for Streamable HTTP.") + + # Don't modify the URL path - respect the exact URL provided by the user + # Some servers are strict about trailing slashes (e.g., PayPal MCP) + + self.url: str = url + self.headers = headers or {} + self.httpx_client_factory = httpx_client_factory + self._set_auth(auth) + + if sse_read_timeout is not None: + if fastmcp.settings.deprecation_warnings: + import warnings + + warnings.warn( + "The `sse_read_timeout` parameter is deprecated and no longer used. " + "The new streamable_http_client API does not support this parameter. " + "Use `read_timeout_seconds` in session_kwargs or configure timeout on " + "the httpx client via `httpx_client_factory` instead.", + DeprecationWarning, + stacklevel=2, + ) + if isinstance(sse_read_timeout, int | float): + sse_read_timeout = datetime.timedelta(seconds=float(sse_read_timeout)) + self.sse_read_timeout = sse_read_timeout + + self._get_session_id_cb: Callable[[], str | None] | None = None + + def _set_auth(self, auth: httpx.Auth | Literal["oauth"] | str | None): + if auth == "oauth": + auth = OAuth(self.url, httpx_client_factory=self.httpx_client_factory) + elif isinstance(auth, str): + auth = BearerAuth(auth) + self.auth = auth + + @contextlib.asynccontextmanager + async def connect_session( + self, **session_kwargs: Unpack[SessionKwargs] + ) -> AsyncIterator[ClientSession]: + # Load headers from an active HTTP request, if available. This will only be true + # if the client is used in a FastMCP Proxy, in which case the MCP client headers + # need to be forwarded to the remote server. + headers = get_http_headers() | self.headers + + # Configure timeout if provided, preserving MCP's 30s connect default + timeout: httpx.Timeout | None = None + if session_kwargs.get("read_timeout_seconds") is not None: + read_timeout_seconds = cast( + datetime.timedelta, session_kwargs.get("read_timeout_seconds") + ) + timeout = httpx.Timeout(30.0, read=read_timeout_seconds.total_seconds()) + + # Create httpx client from factory or use default with MCP-appropriate timeouts + # create_mcp_http_client uses 30s connect/5min read timeout by default, + # and always enables follow_redirects + if self.httpx_client_factory is not None: + # Factory clients get the full kwargs for backwards compatibility + http_client = self.httpx_client_factory( + headers=headers, + auth=self.auth, + follow_redirects=True, # type: ignore[call-arg] + **({"timeout": timeout} if timeout else {}), + ) + else: + http_client = create_mcp_http_client( + headers=headers, + timeout=timeout, + auth=self.auth, + ) + + # Ensure httpx client is closed after use + async with ( + http_client, + streamable_http_client(self.url, http_client=http_client) as transport, + ): + read_stream, write_stream, get_session_id = transport + self._get_session_id_cb = get_session_id + async with ClientSession( + read_stream, write_stream, **session_kwargs + ) as session: + yield session + + def get_session_id(self) -> str | None: + if self._get_session_id_cb: + try: + return self._get_session_id_cb() + except Exception: + return None + return None + + async def close(self): + # Reset the session id callback + self._get_session_id_cb = None + + def __repr__(self) -> str: + return f"" diff --git a/src/fastmcp/client/transports/inference.py b/src/fastmcp/client/transports/inference.py new file mode 100644 index 000000000..de26e37d0 --- /dev/null +++ b/src/fastmcp/client/transports/inference.py @@ -0,0 +1,153 @@ +from pathlib import Path +from typing import TYPE_CHECKING, Any, cast, overload + +from mcp.server.fastmcp import FastMCP as FastMCP1Server +from pydantic import AnyUrl + +from fastmcp.client.transports.base import ClientTransport, ClientTransportT +from fastmcp.client.transports.config import MCPConfigTransport +from fastmcp.client.transports.http import SSETransport, StreamableHttpTransport +from fastmcp.client.transports.memory import FastMCPTransport +from fastmcp.client.transports.stdio import NodeStdioTransport, PythonStdioTransport +from fastmcp.mcp_config import MCPConfig, infer_transport_type_from_url +from fastmcp.server.server import FastMCP +from fastmcp.utilities.logging import get_logger + +if TYPE_CHECKING: + pass + +logger = get_logger(__name__) + + +@overload +def infer_transport(transport: ClientTransportT) -> ClientTransportT: ... + + +@overload +def infer_transport(transport: FastMCP) -> FastMCPTransport: ... + + +@overload +def infer_transport(transport: FastMCP1Server) -> FastMCPTransport: ... + + +@overload +def infer_transport(transport: MCPConfig) -> MCPConfigTransport: ... + + +@overload +def infer_transport(transport: dict[str, Any]) -> MCPConfigTransport: ... + + +@overload +def infer_transport( + transport: AnyUrl, +) -> SSETransport | StreamableHttpTransport: ... + + +@overload +def infer_transport( + transport: str, +) -> ( + PythonStdioTransport | NodeStdioTransport | SSETransport | StreamableHttpTransport +): ... + + +@overload +def infer_transport(transport: Path) -> PythonStdioTransport | NodeStdioTransport: ... + + +def infer_transport( + transport: ClientTransport + | FastMCP + | FastMCP1Server + | AnyUrl + | Path + | MCPConfig + | dict[str, Any] + | str, +) -> ClientTransport: + """ + Infer the appropriate transport type from the given transport argument. + + This function attempts to infer the correct transport type from the provided + argument, handling various input types and converting them to the appropriate + ClientTransport subclass. + + The function supports these input types: + - ClientTransport: Used directly without modification + - FastMCP or FastMCP1Server: Creates an in-memory FastMCPTransport + - Path or str (file path): Creates PythonStdioTransport (.py) or NodeStdioTransport (.js) + - AnyUrl or str (URL): Creates StreamableHttpTransport (default) or SSETransport (for /sse endpoints) + - MCPConfig or dict: Creates MCPConfigTransport, potentially connecting to multiple servers + + For HTTP URLs, they are assumed to be Streamable HTTP URLs unless they end in `/sse`. + + For MCPConfig with multiple servers, a composite client is created where each server + is mounted with its name as prefix. This allows accessing tools and resources from multiple + servers through a single unified client interface, using naming patterns like + `servername_toolname` for tools and `protocol://servername/path` for resources. + If the MCPConfig contains only one server, a direct connection is established without prefixing. + + Examples: + ```python + # Connect to a local Python script + transport = infer_transport("my_script.py") + + # Connect to a remote server via HTTP + transport = infer_transport("http://example.com/mcp") + + # Connect to multiple servers using MCPConfig + config = { + "mcpServers": { + "weather": {"url": "http://weather.example.com/mcp"}, + "calendar": {"url": "http://calendar.example.com/mcp"} + } + } + transport = infer_transport(config) + ``` + """ + + # the transport is already a ClientTransport + if isinstance(transport, ClientTransport): + return transport + + # the transport is a FastMCP server (2.x or 1.0) + elif isinstance(transport, FastMCP | FastMCP1Server): + inferred_transport = FastMCPTransport( + mcp=cast(FastMCP[Any] | FastMCP1Server, transport) + ) + + # the transport is a path to a script + elif isinstance(transport, Path | str) and Path(transport).exists(): + if str(transport).endswith(".py"): + inferred_transport = PythonStdioTransport(script_path=cast(Path, transport)) + elif str(transport).endswith(".js"): + inferred_transport = NodeStdioTransport(script_path=cast(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"): + inferred_transport_type = infer_transport_type_from_url( + cast(AnyUrl | str, transport) + ) + if inferred_transport_type == "sse": + inferred_transport = SSETransport(url=cast(AnyUrl | str, transport)) + else: + inferred_transport = StreamableHttpTransport( + url=cast(AnyUrl | str, transport) + ) + + # if the transport is a config dict or MCPConfig + elif isinstance(transport, dict | MCPConfig): + inferred_transport = MCPConfigTransport( + config=cast(dict | MCPConfig, 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/src/fastmcp/client/transports/memory.py b/src/fastmcp/client/transports/memory.py new file mode 100644 index 000000000..26b4e934a --- /dev/null +++ b/src/fastmcp/client/transports/memory.py @@ -0,0 +1,90 @@ +import contextlib +from collections.abc import AsyncIterator + +import anyio +from mcp import ClientSession +from mcp.server.fastmcp import FastMCP as FastMCP1Server +from mcp.shared.memory import create_client_server_memory_streams +from typing_extensions import Unpack + +from fastmcp.client.transports.base import ClientTransport, SessionKwargs +from fastmcp.server.server import FastMCP + + +class FastMCPTransport(ClientTransport): + """In-memory transport for FastMCP servers. + + This transport connects directly to a FastMCP server instance in the same + Python process. It works with both FastMCP 2.x servers and FastMCP 1.0 + servers from the low-level MCP SDK. This is particularly useful for unit + tests or scenarios where client and server run in the same runtime. + """ + + def __init__(self, mcp: FastMCP | FastMCP1Server, raise_exceptions: bool = False): + """Initialize a FastMCPTransport from a FastMCP server instance.""" + + # Accept both FastMCP 2.x and FastMCP 1.0 servers. Both expose a + # ``_mcp_server`` attribute pointing to the underlying MCP server + # implementation, so we can treat them identically. + self.server = mcp + self.raise_exceptions = raise_exceptions + + @contextlib.asynccontextmanager + async def connect_session( + self, **session_kwargs: Unpack[SessionKwargs] + ) -> AsyncIterator[ClientSession]: + async with create_client_server_memory_streams() as ( + client_streams, + server_streams, + ): + client_read, client_write = client_streams + server_read, server_write = server_streams + + # Capture exceptions to re-raise after task group cleanup. + # anyio task groups can suppress exceptions when cancel_scope.cancel() + # is called during cleanup, so we capture and re-raise manually. + exception_to_raise: BaseException | None = None + + async with ( + anyio.create_task_group() as tg, + _enter_server_lifespan(server=self.server), + ): + tg.start_soon( + lambda: self.server._mcp_server.run( + server_read, + server_write, + self.server._mcp_server.create_initialization_options(), + raise_exceptions=self.raise_exceptions, + ) + ) + + try: + async with ClientSession( + read_stream=client_read, + write_stream=client_write, + **session_kwargs, + ) as client_session: + yield client_session + except BaseException as e: + exception_to_raise = e + finally: + tg.cancel_scope.cancel() + + # Re-raise after task group has exited cleanly + if exception_to_raise is not None: + raise exception_to_raise + + def __repr__(self) -> str: + return f"" + + +@contextlib.asynccontextmanager +async def _enter_server_lifespan( + server: FastMCP | FastMCP1Server, +) -> AsyncIterator[None]: + """Enters the server's lifespan context for FastMCP servers and does nothing for FastMCP 1 servers.""" + if isinstance(server, FastMCP): + async with server._lifespan_manager(): + yield + else: + yield diff --git a/src/fastmcp/client/transports/stdio.py b/src/fastmcp/client/transports/stdio.py new file mode 100644 index 000000000..ca8d5377c --- /dev/null +++ b/src/fastmcp/client/transports/stdio.py @@ -0,0 +1,543 @@ +import asyncio +import contextlib +import os +import shutil +import sys +from collections.abc import AsyncIterator +from pathlib import Path +from typing import TextIO, cast + +import anyio +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client +from typing_extensions import Unpack + +from fastmcp.client.transports.base import ClientTransport, SessionKwargs +from fastmcp.utilities.logging import get_logger +from fastmcp.utilities.mcp_server_config.v1.environments.uv import UVEnvironment + +logger = get_logger(__name__) + + +class StdioTransport(ClientTransport): + """ + Base transport for connecting to an MCP server via subprocess with stdio. + + This is a base class that can be subclassed for specific command-based + transports like Python, Node, Uvx, etc. + """ + + def __init__( + self, + command: str, + args: list[str], + env: dict[str, str] | None = None, + cwd: str | None = None, + keep_alive: bool | None = None, + log_file: Path | TextIO | None = None, + ): + """ + Initialize a Stdio transport. + + Args: + command: The command to run (e.g., "python", "node", "uvx") + args: The arguments to pass to the command + env: Environment variables to set for the subprocess + cwd: Current working directory for the subprocess + keep_alive: Whether to keep the subprocess alive between connections. + Defaults to True. When True, the subprocess remains active + after the connection context exits, allowing reuse in + subsequent connections. + log_file: Optional path or file-like object where subprocess stderr will + be written. Can be a Path or TextIO object. Defaults to sys.stderr + if not provided. When a Path is provided, the file will be created + if it doesn't exist, or appended to if it does. When set, server + errors will be written to this file instead of appearing in the console. + """ + self.command = command + self.args = args + self.env = env + self.cwd = cwd + if keep_alive is None: + keep_alive = True + self.keep_alive = keep_alive + self.log_file = log_file + + self._session: ClientSession | None = None + self._connect_task: asyncio.Task | None = None + self._ready_event = anyio.Event() + self._stop_event = anyio.Event() + + @contextlib.asynccontextmanager + async def connect_session( + self, **session_kwargs: Unpack[SessionKwargs] + ) -> AsyncIterator[ClientSession]: + try: + await self.connect(**session_kwargs) + yield cast(ClientSession, self._session) + finally: + if not self.keep_alive: + await self.disconnect() + else: + logger.debug("Stdio transport has keep_alive=True, not disconnecting") + + async def connect( + self, **session_kwargs: Unpack[SessionKwargs] + ) -> ClientSession | None: + if self._connect_task is not None: + return + + session_future: asyncio.Future[ClientSession] = asyncio.Future() + + # start the connection task + self._connect_task = asyncio.create_task( + _stdio_transport_connect_task( + command=self.command, + args=self.args, + env=self.env, + cwd=self.cwd, + log_file=self.log_file, + # TODO(ty): remove when ty supports Unpack[TypedDict] inference + session_kwargs=session_kwargs, # type: ignore[arg-type] + ready_event=self._ready_event, + stop_event=self._stop_event, + session_future=session_future, + ) + ) + + # wait for the client to be ready before returning + await self._ready_event.wait() + + # Check if connect task completed with an exception (early failure) + if self._connect_task.done(): + exception = self._connect_task.exception() + if exception is not None: + raise exception + + self._session = await session_future + return self._session + + async def disconnect(self): + if self._connect_task is None: + return + + # signal the connection task to stop + self._stop_event.set() + + # wait for the connection task to finish cleanly + await self._connect_task + + # reset variables and events for potential future reconnects + self._connect_task = None + self._stop_event = anyio.Event() + self._ready_event = anyio.Event() + + async def close(self): + await self.disconnect() + + def __del__(self): + """Ensure that we send a disconnection signal to the transport task if we are being garbage collected.""" + if not self._stop_event.is_set(): + self._stop_event.set() + + def __repr__(self) -> str: + return ( + f"<{self.__class__.__name__}(command='{self.command}', args={self.args})>" + ) + + +async def _stdio_transport_connect_task( + command: str, + args: list[str], + env: dict[str, str] | None, + cwd: str | None, + log_file: Path | TextIO | None, + session_kwargs: SessionKwargs, + ready_event: anyio.Event, + stop_event: anyio.Event, + session_future: asyncio.Future[ClientSession], +): + """A standalone connection task for a stdio transport. It is not a part of the StdioTransport class + to ensure that the connection task does not hold a reference to the Transport object.""" + + try: + async with contextlib.AsyncExitStack() as stack: + try: + server_params = StdioServerParameters( + command=command, + args=args, + env=env, + cwd=cwd, + ) + # Handle log_file: Path needs to be opened, TextIO used as-is + if log_file is None: + log_file_handle = sys.stderr + elif isinstance(log_file, Path): + log_file_handle = stack.enter_context(log_file.open("a")) + else: + # Must be TextIO - use it directly + log_file_handle = log_file + + transport = await stack.enter_async_context( + stdio_client(server_params, errlog=log_file_handle) + ) + read_stream, write_stream = transport + session_future.set_result( + await stack.enter_async_context( + ClientSession(read_stream, write_stream, **session_kwargs) + ) + ) + + logger.debug("Stdio transport connected") + ready_event.set() + + # Wait until disconnect is requested (stop_event is set) + await stop_event.wait() + finally: + # Clean up client on exit + logger.debug("Stdio transport disconnected") + except Exception: + # Ensure ready event is set even if connection fails + ready_event.set() + raise + + +class PythonStdioTransport(StdioTransport): + """Transport for running Python scripts.""" + + def __init__( + self, + script_path: str | Path, + args: list[str] | None = None, + env: dict[str, str] | None = None, + cwd: str | None = None, + python_cmd: str = sys.executable, + keep_alive: bool | None = None, + log_file: Path | TextIO | None = None, + ): + """ + Initialize a Python transport. + + Args: + script_path: Path to the Python script to run + args: Additional arguments to pass to the script + env: Environment variables to set for the subprocess + cwd: Current working directory for the subprocess + python_cmd: Python command to use (default: "python") + keep_alive: Whether to keep the subprocess alive between connections. + Defaults to True. When True, the subprocess remains active + after the connection context exits, allowing reuse in + subsequent connections. + log_file: Optional path or file-like object where subprocess stderr will + be written. Can be a Path or TextIO object. Defaults to sys.stderr + if not provided. When a Path is provided, the file will be created + if it doesn't exist, or appended to if it does. When set, server + errors will be written to this file instead of appearing in the console. + """ + script_path = Path(script_path).resolve() + if not script_path.is_file(): + raise FileNotFoundError(f"Script not found: {script_path}") + if not str(script_path).endswith(".py"): + raise ValueError(f"Not a Python script: {script_path}") + + full_args = [str(script_path)] + if args: + full_args.extend(args) + + super().__init__( + command=python_cmd, + args=full_args, + env=env, + cwd=cwd, + keep_alive=keep_alive, + log_file=log_file, + ) + self.script_path = script_path + + +class FastMCPStdioTransport(StdioTransport): + """Transport for running FastMCP servers using the FastMCP CLI.""" + + def __init__( + self, + script_path: str | Path, + args: list[str] | None = None, + env: dict[str, str] | None = None, + cwd: str | None = None, + keep_alive: bool | None = None, + log_file: Path | TextIO | None = None, + ): + script_path = Path(script_path).resolve() + if not script_path.is_file(): + raise FileNotFoundError(f"Script not found: {script_path}") + if not str(script_path).endswith(".py"): + raise ValueError(f"Not a Python script: {script_path}") + + super().__init__( + command="fastmcp", + args=["run", str(script_path)], + env=env, + cwd=cwd, + keep_alive=keep_alive, + log_file=log_file, + ) + self.script_path = script_path + + +class NodeStdioTransport(StdioTransport): + """Transport for running Node.js scripts.""" + + def __init__( + self, + script_path: str | Path, + args: list[str] | None = None, + env: dict[str, str] | None = None, + cwd: str | None = None, + node_cmd: str = "node", + keep_alive: bool | None = None, + log_file: Path | TextIO | None = None, + ): + """ + Initialize a Node transport. + + Args: + script_path: Path to the Node.js script to run + args: Additional arguments to pass to the script + env: Environment variables to set for the subprocess + cwd: Current working directory for the subprocess + node_cmd: Node.js command to use (default: "node") + keep_alive: Whether to keep the subprocess alive between connections. + Defaults to True. When True, the subprocess remains active + after the connection context exits, allowing reuse in + subsequent connections. + log_file: Optional path or file-like object where subprocess stderr will + be written. Can be a Path or TextIO object. Defaults to sys.stderr + if not provided. When a Path is provided, the file will be created + if it doesn't exist, or appended to if it does. When set, server + errors will be written to this file instead of appearing in the console. + """ + script_path = Path(script_path).resolve() + if not script_path.is_file(): + raise FileNotFoundError(f"Script not found: {script_path}") + if not str(script_path).endswith(".js"): + raise ValueError(f"Not a JavaScript script: {script_path}") + + full_args = [str(script_path)] + if args: + full_args.extend(args) + + super().__init__( + command=node_cmd, + args=full_args, + env=env, + cwd=cwd, + keep_alive=keep_alive, + log_file=log_file, + ) + self.script_path = script_path + + +class UvStdioTransport(StdioTransport): + """Transport for running commands via the uv tool.""" + + def __init__( + self, + command: str, + args: list[str] | None = None, + module: bool = False, + project_directory: Path | None = None, + python_version: str | None = None, + with_packages: list[str] | None = None, + with_requirements: Path | None = None, + env_vars: dict[str, str] | None = None, + keep_alive: bool | None = None, + ): + # Basic validation + if project_directory and not project_directory.exists(): + raise NotADirectoryError( + f"Project directory not found: {project_directory}" + ) + + # Create Environment from provided parameters (internal use) + env_config = UVEnvironment( + python=python_version, + dependencies=with_packages, + requirements=with_requirements, + project=project_directory, + editable=None, # Not exposed in this transport + ) + + # Build uv arguments using the config + uv_args: list[str] = [] + + # Check if we need any environment setup + if env_config._must_run_with_uv(): + # Use the config to build args, but we need to handle the command differently + # since transport has specific needs + uv_args = ["run"] + + if python_version: + uv_args.extend(["--python", python_version]) + if project_directory: + uv_args.extend(["--directory", str(project_directory)]) + + # Note: Don't add fastmcp as dependency here, transport is for general use + for pkg in with_packages or []: + uv_args.extend(["--with", pkg]) + if with_requirements: + uv_args.extend(["--with-requirements", str(with_requirements)]) + else: + # No environment setup needed + uv_args = ["run"] + + if module: + uv_args.append("--module") + + if not args: + args = [] + + uv_args.extend([command, *args]) + + # Get environment with any additional variables + env: dict[str, str] | None = None + if env_vars or project_directory: + env = os.environ.copy() + if project_directory: + env["UV_PROJECT_DIR"] = str(project_directory) + if env_vars: + env.update(env_vars) + + super().__init__( + command="uv", + args=uv_args, + env=env, + cwd=None, # Use --directory flag instead of cwd + keep_alive=keep_alive, + ) + + +class UvxStdioTransport(StdioTransport): + """Transport for running commands via the uvx tool.""" + + def __init__( + self, + tool_name: str, + tool_args: list[str] | None = None, + project_directory: str | None = None, + python_version: str | None = None, + with_packages: list[str] | None = None, + from_package: str | None = None, + env_vars: dict[str, str] | None = None, + keep_alive: bool | None = None, + ): + """ + Initialize a Uvx transport. + + Args: + tool_name: Name of the tool to run via uvx + tool_args: Arguments to pass to the tool + project_directory: Project directory (for package resolution) + python_version: Python version to use + with_packages: Additional packages to include + from_package: Package to install the tool from + env_vars: Additional environment variables + keep_alive: Whether to keep the subprocess alive between connections. + Defaults to True. When True, the subprocess remains active + after the connection context exits, allowing reuse in + subsequent connections. + """ + # Basic validation + if project_directory and not Path(project_directory).exists(): + raise NotADirectoryError( + f"Project directory not found: {project_directory}" + ) + + # Build uvx arguments + uvx_args: list[str] = [] + if python_version: + uvx_args.extend(["--python", python_version]) + if from_package: + uvx_args.extend(["--from", from_package]) + for pkg in with_packages or []: + uvx_args.extend(["--with", pkg]) + + # Add the tool name and tool args + uvx_args.append(tool_name) + if tool_args: + uvx_args.extend(tool_args) + + env: dict[str, str] | None = None + if env_vars: + env = os.environ.copy() + env.update(env_vars) + + super().__init__( + command="uvx", + args=uvx_args, + env=env, + cwd=project_directory, + keep_alive=keep_alive, + ) + self.tool_name: str = tool_name + + +class NpxStdioTransport(StdioTransport): + """Transport for running commands via the npx tool.""" + + def __init__( + self, + package: str, + args: list[str] | None = None, + project_directory: str | None = None, + env_vars: dict[str, str] | None = None, + use_package_lock: bool = True, + keep_alive: bool | None = None, + ): + """ + Initialize an Npx transport. + + Args: + package: Name of the npm package to run + args: Arguments to pass to the package command + project_directory: Project directory with package.json + env_vars: Additional environment variables + use_package_lock: Whether to use package-lock.json (--prefer-offline) + keep_alive: Whether to keep the subprocess alive between connections. + Defaults to True. When True, the subprocess remains active + after the connection context exits, allowing reuse in + subsequent connections. + """ + # verify npx is installed + if shutil.which("npx") is None: + raise ValueError("Command 'npx' not found") + + # Basic validation + if project_directory and not Path(project_directory).exists(): + raise NotADirectoryError( + f"Project directory not found: {project_directory}" + ) + + # Build npx arguments + npx_args = [] + if use_package_lock: + npx_args.append("--prefer-offline") + + # Add the package name and args + npx_args.append(package) + if args: + npx_args.extend(args) + + # Get environment with any additional variables + env = None + if env_vars: + env = os.environ.copy() + env.update(env_vars) + + super().__init__( + command="npx", + args=npx_args, + env=env, + cwd=project_directory, + keep_alive=keep_alive, + ) + self.package = package