Merge pull request #522 from jlowin/config-dicts

src/fastmcp/client/transports.py

@@ -6,8 +6,7 @@ import shutil
 import sys
 from collections.abc import AsyncIterator
 from pathlib import Path
-from typing import Any, TypedDict, cast
-from urllib.parse import urlparse
+from typing import TYPE_CHECKING, Any, TypedDict, cast
 
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.session import (
@@ -26,6 +25,10 @@ from typing_extensions import Unpack
 
 from fastmcp.server import FastMCP as FastMCPServer
 from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.mcp_config import MCPConfig, infer_transport_type_from_url
+
+if TYPE_CHECKING:
+    from fastmcp.utilities.mcp_config import MCPConfig
 
 logger = get_logger(__name__)
 
@@ -470,7 +473,13 @@ class FastMCPTransport(ClientTransport):
 
 
 def infer_transport(
-    transport: ClientTransport | FastMCPServer | AnyUrl | Path | dict[str, Any] | str,
+    transport: ClientTransport
+    | FastMCPServer
+    | AnyUrl
+    | Path
+    | MCPConfig
+    | dict[str, Any]
+    | str,
 ) -> ClientTransport:
     """
     Infer the appropriate transport type from the given transport argument.
@@ -481,6 +490,8 @@ def infer_transport(
 
     For HTTP URLs, they are assumed to be Streamable HTTP URLs unless they end in `/sse`.
     """
+    from fastmcp.utilities.mcp_config import MCPConfig
+
     # the transport is already a ClientTransport
     if isinstance(transport, ClientTransport):
         return transport
@@ -500,45 +511,24 @@ def infer_transport(
 
     # the transport is an http(s) URL
     elif isinstance(transport, AnyUrl | str) and str(transport).startswith("http"):
-        transport_str = str(transport)
-        # Parse out just the path portion to check for /sse
-        parsed_url = urlparse(transport_str)
-        path = parsed_url.path
-
-        # Check if path contains /sse/ or ends with /sse
-        if "/sse/" in path or path.rstrip("/").endswith("/sse"):
+        inferred_transport_type = infer_transport_type_from_url(transport)
+        if inferred_transport_type == "sse":
             inferred_transport = SSETransport(url=transport)
         else:
             inferred_transport = StreamableHttpTransport(url=transport)
 
-    ## if the transport is a config dict
-    elif isinstance(transport, dict):
-        if "mcpServers" not in transport:
-            raise ValueError("Invalid transport dictionary: missing 'mcpServers' key")
+    # if the transport is a config dict or MCPConfig
+    elif isinstance(transport, dict | MCPConfig):
+        if isinstance(transport, dict):
+            config = MCPConfig.from_dict(transport)
         else:
-            server = transport["mcpServers"]
-            if len(list(server.keys())) > 1:
-                raise ValueError(
-                    "Invalid transport dictionary: multiple servers found - only one expected"
-                )
-            server_name = list(server.keys())[0]
-            # Stdio transport
-            if "command" in server[server_name] and "args" in server[server_name]:
-                inferred_transport = StdioTransport(
-                    command=server[server_name]["command"],
-                    args=server[server_name]["args"],
-                    env=server[server_name].get("env", None),
-                    cwd=server[server_name].get("cwd", None),
-                )
-
-            # HTTP transport
-            elif "url" in server:
-                inferred_transport = SSETransport(
-                    url=server["url"],
-                    headers=server.get("headers", None),
-                )
-
-            raise ValueError("Cannot determine transport type from dictionary")
+            config = transport
+        inferred_transports = config.to_transports()
+        if len(inferred_transports) > 1:
+            raise ValueError(
+                "Invalid transport dictionary: multiple servers found - only one expected"
+            )
+        inferred_transport = list(inferred_transports.values())[0]
 
     # the transport is an unknown type
     else:

src/fastmcp/server/proxy.py

@@ -25,6 +25,7 @@ from fastmcp.server.context import Context
 from fastmcp.server.server import FastMCP
 from fastmcp.tools.tool import Tool
 from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.mcp_config import MCPConfig
 
 if TYPE_CHECKING:
     from fastmcp.server import Context
@@ -177,6 +178,13 @@ class FastMCPProxy(FastMCP):
         super().__init__(**kwargs)
         self.client = client
 
+    @classmethod
+    async def from_mcp_config(cls, config: MCPConfig | dict) -> FastMCPProxy:
+        if isinstance(config, dict):
+            config = MCPConfig.from_dict(config)
+        clients = config.to_clients()
+        return cls(client=clients[list(clients.keys())[0]])
+
     async def get_tools(self) -> dict[str, Tool]:
         tools = await super().get_tools()
 

src/fastmcp/utilities/mcp_config.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Literal
+from urllib.parse import urlparse
+
+from pydantic import AnyUrl, BaseModel, Field
+
+if TYPE_CHECKING:
+    from fastmcp.client.client import Client
+    from fastmcp.client.transports import (
+        SSETransport,
+        StdioTransport,
+        StreamableHttpTransport,
+    )
+
+
+def infer_transport_type_from_url(
+    url: str | AnyUrl,
+) -> Literal["streamable-http", "sse"]:
+    """
+    Infer the appropriate transport type from the given URL.
+    """
+    url = str(url)
+    if not url.startswith("http"):
+        raise ValueError(f"Invalid URL: {url}")
+
+    parsed_url = urlparse(url)
+    path = parsed_url.path
+
+    if "/sse/" in path or path.rstrip("/").endswith("/sse"):
+        return "sse"
+    else:
+        return "streamable-http"
+
+
+class LocalMCPServer(BaseModel):
+    command: str
+    args: list[str] = Field(default_factory=list)
+    env: dict[str, Any] = Field(default_factory=dict)
+    cwd: str | None = None
+
+    def to_transport(self) -> StdioTransport:
+        from fastmcp.client.transports import StdioTransport
+
+        return StdioTransport(
+            command=self.command,
+            args=self.args,
+            env=self.env,
+            cwd=self.cwd,
+        )
+
+
+class RemoteMCPServer(BaseModel):
+    url: str
+    transport: Literal["streamable-http", "sse", "http"] | None = None
+    headers: dict[str, str] = Field(default_factory=dict)
+
+    def to_transport(self) -> StreamableHttpTransport | SSETransport:
+        from fastmcp.client.transports import SSETransport, StreamableHttpTransport
+
+        if self.transport is None:
+            transport = infer_transport_type_from_url(self.url)
+        else:
+            transport = self.transport
+
+        if transport == "sse":
+            return SSETransport(self.url, headers=self.headers)
+        else:
+            return StreamableHttpTransport(self.url, headers=self.headers)
+
+
+class MCPConfig(BaseModel):
+    mcpServers: dict[str, LocalMCPServer | RemoteMCPServer]
+
+    @classmethod
+    def from_dict(cls, config: dict[str, Any]) -> MCPConfig:
+        return cls(mcpServers=config.get("mcpServers", config))
+
+    def to_transports(
+        self,
+    ) -> dict[str, StdioTransport | StreamableHttpTransport | SSETransport]:
+        return {name: server.to_transport() for name, server in self.mcpServers.items()}
+
+    def to_clients(self) -> dict[str, Client]:
+        from fastmcp.client.client import Client
+
+        return {
+            name: Client(transport=transport)
+            for name, transport in self.to_transports().items()
+        }

tests/client/test_client.py

@@ -10,6 +10,7 @@ from fastmcp.client import Client
 from fastmcp.client.transports import (
     FastMCPTransport,
     SSETransport,
+    StdioTransport,
     StreamableHttpTransport,
     infer_transport,
 )
@@ -640,3 +641,31 @@ class TestInferTransport:
     def test_url_returns_streamable_http_transport(self, url):
         """Test that URLs without /sse/ pattern return StreamableHttpTransport."""
         assert isinstance(infer_transport(url), StreamableHttpTransport)
+
+    def test_infer_remote_transport_from_config(self):
+        config = {
+            "mcpServers": {
+                "test_server": {
+                    "url": "http://localhost:8000/sse",
+                    "headers": {"Authorization": "Bearer 123"},
+                },
+            }
+        }
+        transport = infer_transport(config)
+        assert isinstance(transport, SSETransport)
+        assert transport.url == "http://localhost:8000/sse"
+        assert transport.headers == {"Authorization": "Bearer 123"}
+
+    def test_infer_local_transport_from_config(self):
+        config = {
+            "mcpServers": {
+                "test_server": {
+                    "command": "echo",
+                    "args": ["hello"],
+                },
+            }
+        }
+        transport = infer_transport(config)
+        assert isinstance(transport, StdioTransport)
+        assert transport.command == "echo"
+        assert transport.args == ["hello"]

tests/utilities/test_mcp_config.py

@@ -0,0 +1,92 @@
+from fastmcp.client.transports import (
+    SSETransport,
+    StdioTransport,
+    StreamableHttpTransport,
+)
+from fastmcp.utilities.mcp_config import LocalMCPServer, MCPConfig, RemoteMCPServer
+
+
+def test_parse_single_stdio_config():
+    config = {
+        "mcpServers": {
+            "test_server": {
+                "command": "echo",
+                "args": ["hello"],
+            }
+        }
+    }
+    mcp_config = MCPConfig.from_dict(config)
+    transport = mcp_config.mcpServers["test_server"].to_transport()
+    assert isinstance(transport, StdioTransport)
+    assert transport.command == "echo"
+    assert transport.args == ["hello"]
+
+
+def test_parse_single_remote_config():
+    config = {
+        "mcpServers": {
+            "test_server": {
+                "url": "http://localhost:8000",
+            }
+        }
+    }
+    mcp_config = MCPConfig.from_dict(config)
+    transport = mcp_config.mcpServers["test_server"].to_transport()
+    assert isinstance(transport, StreamableHttpTransport)
+    assert transport.url == "http://localhost:8000"
+
+
+def test_parse_remote_config_with_transport():
+    config = {
+        "mcpServers": {
+            "test_server": {
+                "url": "http://localhost:8000",
+                "transport": "sse",
+            }
+        }
+    }
+    mcp_config = MCPConfig.from_dict(config)
+    transport = mcp_config.mcpServers["test_server"].to_transport()
+    assert isinstance(transport, SSETransport)
+    assert transport.url == "http://localhost:8000"
+
+
+def test_parse_remote_config_with_url_inference():
+    config = {
+        "mcpServers": {
+            "test_server": {
+                "url": "http://localhost:8000/sse",
+            }
+        }
+    }
+    mcp_config = MCPConfig.from_dict(config)
+    transport = mcp_config.mcpServers["test_server"].to_transport()
+    assert isinstance(transport, SSETransport)
+    assert transport.url == "http://localhost:8000/sse"
+
+
+def test_parse_multiple_servers():
+    config = {
+        "mcpServers": {
+            "test_server": {
+                "url": "http://localhost:8000/sse",
+            },
+            "test_server_2": {
+                "command": "echo",
+                "args": ["hello"],
+                "env": {"TEST": "test"},
+            },
+        }
+    }
+    mcp_config = MCPConfig.from_dict(config)
+    assert len(mcp_config.mcpServers) == 2
+    assert isinstance(mcp_config.mcpServers["test_server"], RemoteMCPServer)
+    assert isinstance(mcp_config.mcpServers["test_server"].to_transport(), SSETransport)
+
+    assert isinstance(mcp_config.mcpServers["test_server_2"], LocalMCPServer)
+    assert isinstance(
+        mcp_config.mcpServers["test_server_2"].to_transport(), StdioTransport
+    )
+    assert mcp_config.mcpServers["test_server_2"].command == "echo"
+    assert mcp_config.mcpServers["test_server_2"].args == ["hello"]
+    assert mcp_config.mcpServers["test_server_2"].env == {"TEST": "test"}