New in version: 2.0.0

FastMCP Client依赖于ClientTransport对象来处理与MCP服务器连接和通信的具体细节。FastMCP为常见的连接方式提供了多种内置传输实现。

虽然Client通常会自动推断正确的传输方式（参见Client Overview），但您也可以显式实例化传输以获得更多控制权。

客户端是轻量级对象，因此可以根据需要随时创建新实例。但请注意上下文管理——每次打开客户端上下文(async with client:)时，都会建立新的连接或进程。为了获得最佳性能，在执行多个操作时应保持客户端上下文处于打开状态，而不是反复开关它们。

​
选择传输方式

选择最适合您使用场景的传输方式：

  • 连接远程/持久化服务器： 对于基于Web的部署，推荐使用StreamableHttpTransport（HTTP URL的默认选项）或SSETransport（传统选项）。

  • 本地开发/测试： 使用 FastMCPTransport 进行内存中的同进程测试，用于测试您的 FastMCP 服务器。

  • 运行本地服务器： 如果需要将MCP服务器作为打包工具运行，可以使用UvxStdioTransport (Python/uv) 或 NpxStdioTransport (Node/npm)。

​
网络传输

这些传输方式连接到运行在网络上的服务器，通常是可通过URL访问的长期运行服务。

​
可流式传输的HTTP

New in version: 2.3.0

可流式HTTP是基于Web部署场景的推荐传输协议，能通过HTTP实现高效的双向通信。

​
概述

  • 类: fastmcp.client.transports.StreamableHttpTransport
  • 推断来源： 以 http:// 或 https:// 开头的URL（自v2.3.0起成为HTTP URL的默认设置）
  • 服务器兼容性： 适用于运行在streamable-http模式的FastMCP服务器

​
基本用法

使用Streamable HTTP的最简单方法是让传输方式从URL中自动推断：

from fastmcp import Client
import asyncio

# The Client automatically uses StreamableHttpTransport for HTTP URLs
client = Client("https://example.com/mcp")

async def main():
    async with client:
        tools = await client.list_tools()
        print(f"Available tools: {tools}")

asyncio.run(main())

​
使用请求头认证

对于需要身份验证的服务器：

from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport

# Create transport with authentication headers
transport = StreamableHttpTransport(
    url="https://example.com/mcp",
    headers={"Authorization": "Bearer your-token-here"}
)

client = Client(transport)

​
SSE (服务器推送事件)

New in version: 2.0.0

服务器发送事件(SSE)是一种允许服务器通过HTTP连接向客户端推送数据的传输协议。虽然仍受支持，但对于新的基于Web的部署，现在推荐使用可流式HTTP作为传输协议。

​
概述

  • 类： fastmcp.client.transports.SSETransport
  • 推断来源：自v2.3.0版本起，HTTP URL不再自动推断（必须明确指定）
  • 服务器兼容性： 可与运行在sse模式的FastMCP服务器配合使用

​
基本用法

自v2.3.0版本起，您必须显式地为SSE连接创建一个SSETransport：

from fastmcp import Client
from fastmcp.client.transports import SSETransport
import asyncio

# Create an SSE transport
transport = SSETransport(url="https://example.com/sse")

# Pass the transport to the client
client = Client(transport)

async def main():
    async with client:
        tools = await client.list_tools()
        print(f"Available tools: {tools}")

asyncio.run(main())

​
使用请求头进行身份验证

SSE传输还支持用于身份验证的自定义头部：

from fastmcp import Client
from fastmcp.client.transports import SSETransport

# Create SSE transport with authentication headers
transport = SSETransport(
    url="https://example.com/sse",
    headers={"Authorization": "Bearer your-token-here"}
)

client = Client(transport)

​
何时使用SSE与可流式HTTP

  • 在以下情况下使用可流式HTTP：

    • 设置新部署（推荐使用默认设置）
    • 你需要双向流式传输
    • 您正在连接到运行在streamable-http模式下的FastMCP服务器

  • 在以下情况下使用SSE：

    • 连接到运行在sse模式下的旧版FastMCP服务器
    • 使用专为服务器发送事件优化的基础设施

​
本地传输

这些传输管理器管理着一个作为子进程运行的MCP服务器，通过标准输入(stdin)和标准输出(stdout)与之通信。这是像Claude Desktop这样的客户端使用的标准机制。

​
Python标准输入输出

  • 类: fastmcp.client.transports.PythonStdioTransport
  • 推断来源： 指向 .py 文件的路径
  • 使用场景： 在子进程中运行基于Python的MCP服务器脚本

这是在开发过程中或与期望启动服务器脚本的工具集成时，与本地FastMCP服务器交互的最常见方式。

from fastmcp import Client
from fastmcp.client.transports import PythonStdioTransport

server_script = "my_mcp_server.py" # Path to your server script

# Option 1: Inferred transport
client = Client(server_script)

# Option 2: Explicit transport with custom configuration
transport = PythonStdioTransport(
    script_path=server_script,
    python_cmd="/usr/bin/python3.11", # Optional: specify Python interpreter
    # args=["--some-server-arg"],      # Optional: pass arguments to the script
    # env={"MY_VAR": "value"},         # Optional: set environment variables
)
client = Client(transport)

async def main():
    async with client:
        tools = await client.list_tools()
        print(f"Connected via Python Stdio, found tools: {tools}")

asyncio.run(main())

服务器脚本必须包含启动MCP服务器并通过标准输入输出监听的逻辑，通常通过mcp.run()或fastmcp.server.run()实现。客户端仅负责启动脚本，不会注入服务器逻辑。

​
Node.js 标准输入输出

  • 类： fastmcp.client.transports.NodeStdioTransport
  • 推断来源： 指向 .js 文件的路径
  • 使用场景： 在子进程中运行基于Node.js的MCP服务器脚本

类似于Python传输，但适用于JavaScript服务器。

from fastmcp import Client
from fastmcp.client.transports import NodeStdioTransport

node_server_script = "my_mcp_server.js" # Path to your Node.js server script

# Option 1: Inferred transport
client = Client(node_server_script)

# Option 2: Explicit transport
transport = NodeStdioTransport(
    script_path=node_server_script,
    node_cmd="node" # Optional: specify path to Node executable
)
client = Client(transport)

async def main():
    async with client:
        tools = await client.list_tools()
        print(f"Connected via Node.js Stdio, found tools: {tools}")

asyncio.run(main())

​
UVX 集成开发环境(实验版)

  • 类： fastmcp.client.transports.UvxStdioTransport
  • 推断来源： 非自动推断
  • 使用场景： 使用uvx运行打包为Python工具的MCP服务器

这对于执行以命令行工具或软件包形式分发的MCP服务器非常有用，无需将它们安装到您的环境中。

from fastmcp import Client
from fastmcp.client.transports import UvxStdioTransport

# Run a hypothetical 'cloud-analyzer-mcp' tool via uvx
transport = UvxStdioTransport(
    tool_name="cloud-analyzer-mcp",
    # from_package="cloud-analyzer-cli", # Optional: specify package if tool name differs
    # with_packages=["boto3", "requests"] # Optional: add dependencies
)
client = Client(transport)

async def main():
    async with client:
        result = await client.call_tool("analyze_bucket", {"name": "my-data"})
        print(f"Analysis result: {result}")

asyncio.run(main())

​
NPX 标准输入输出（实验性功能）

  • 类: fastmcp.client.transports.NpxStdioTransport
  • 推断来源： 非自动推断
  • 使用场景：使用npx运行打包为NPM包的MCP服务器

类似于UvxStdioTransport，但适用于Node.js生态系统。

from fastmcp import Client
from fastmcp.client.transports import NpxStdioTransport

# Run an MCP server from an NPM package
transport = NpxStdioTransport(
    package="mcp-server-package",
    # args=["--port", "stdio"] # Optional: pass arguments to the package
)
client = Client(transport)

async def main():
    async with client:
        result = await client.call_tool("get_npm_data", {})
        print(f"Result: {result}")

asyncio.run(main())

​
内存传输

​
FastMCP传输

  • 类: fastmcp.client.transports.FastMCPTransport
  • 推断来源： 一个 fastmcp.server.FastMCP 的实例
  • 使用场景： 在同一Python进程中直接连接到FastMCP服务器实例

这对测试你的FastMCP服务器非常有用。

from fastmcp import FastMCP, Client
import asyncio

# 1. Create your FastMCP server instance
server = FastMCP(name="InMemoryServer")

@server.tool()
def ping(): 
    return "pong"

# 2. Create a client pointing directly to the server instance
client = Client(server)  # Transport is automatically inferred

async def main():
    async with client:
        result = await client.call_tool("ping")
        print(f"In-memory call result: {result}")

asyncio.run(main())

通信通过高效的内存队列进行，速度极快，非常适合单元测试。