fastmcp客户端传输方式代码实战

前言

FastMCP Client 通过处理底层连接机制的传输对象与 MCP 服务器通信。虽然客户端可以根据您传递给它的内容自动选择传输方式，但显式实例化传输可以让您完全控制配置——环境变量、身份验证、会话管理等等。
将传输视为您的客户端代码和 MCP 服务器之间的可配置适配器。每种传输类型处理不同的通信模式：带管道的子进程、HTTP 连接或直接内存调用。

本博客为了完整性，把官网的很多内容都写进来了，如果我们仅仅是想要直接运行几种不同传输方式的代码，请直接参考本文后记的部分。

统一测试的MCP Server

这里就简单的写一个MCP Server供下面不同的MCP Client使用

fromfastmcpimportFastMCP# 初始化FastMCP Servermcp=FastMCP("My MCP Server")# 用@mcp.tool()装饰器定义一个工具@mcp.tool()defadd(a:int,b:int)->int:""" :param a: 第一个整数 :param b: 第二个整数 :return: 返回两个数字之和 """returna+b# 启动MCP Serverif__name__=="__main__":# ssemcp.run(transport="sse",host="127.0.0.1",port=8001)# streamable-http# mcp.run(transport="streamable-http", host="127.0.0.1", port=8001)# stdio# mcp.run(transport="stdio")

选择合适的传输方式

• 使用 STDIO 传输当您需要运行本地 MCP 服务器并完全控制其环境和生命周期时
• 使用 远程传输当连接到独立运行的生产服务或共享 MCP 服务器时
• 使用 内存传输当测试 FastMCP 服务器而不需要子进程或网络开销时
• 使用 MCP JSON 配置当您需要连接到配置文件中定义的多个服务器时

STDIO 传输

STDIO（标准输入/输出）传输通过子进程管道与 MCP 服务器通信。这是桌面客户端（如 Claude Desktop）使用的标准机制，也是运行本地 MCP 服务器的主要方式。

客户端运行服务器

关键概念：使用 STDIO 传输时，您的客户端实际上会启动并管理服务器进程。这与网络传输根本不同，网络传输是连接到已经运行的服务器。理解这种关系是有效使用 STDIO 的关键。

使用 STDIO 传输，您的客户端会：

• 在连接时将服务器作为子进程启动
• 管理服务器的生命周期（启动、停止、重启）
• 控制服务器的环境和配置
• 通过 stdin/stdout 管道进行通信

这种架构支持强大的本地集成，但需要理解环境隔离和进程管理。

环境隔离

STDIO 服务器默认在隔离环境中运行。这是 MCP 协议强制执行的安全功能，用于防止敏感数据的意外暴露。
当您的客户端启动 MCP 服务器时：

• 服务器不会继承您 shell 的环境变量
• API 密钥、路径和其他配置必须显式传递
• 工作目录和系统路径可能与您的 shell 不同

要向服务器传递环境变量，请使用 env 参数：

fromfastmcpimportClient# 如果您的服务器需要环境变量（如 API 密钥），# 您必须显式传递它们：client=Client("my_server.py",env={"API_KEY":"secret","DEBUG":"true"})# 这样不会起作用 - 服务器在隔离环境中运行：# export API_KEY="secret" # 在您的 shell 中# client = Client("my_server.py") # 服务器看不到 API_KEY

基本用法

要使用 STDIO 传输，您需要创建一个传输实例，包含运行服务器所需的命令和参数：

fromfastmcpimportClientfromfastmcp.client.transportsimportStdioTransport transport=StdioTransport(command="python",args=["my_server.py"])asyncdefmain():asyncwithClient(transport=transport)asclient:# 列出可用工具tools=awaitclient.list_tools()print(f"可用的工具有：{tools}")# 调用add工具result=awaitclient.call_tool("add",{"a":1,"b":3})print(f"结果为：{result}")importasyncio asyncio.run(main())

用stdio的方式启动MCP Server后，运行上面的MCP Client代码结果：

[12/20/25 13:55:45] INFO Starting MCP server 'My MCP Server' server.py:2527 with transport 'stdio' 可用的工具有：[Tool(name='add', title=None, description=':param a: 第一个整数\n:param b: 第二个整数\n:return: 返回两个数字之和', inputSchema={'properties': {'a': {'type': 'integer'}, 'b': {'type': 'integer'}}, 'required': ['a', 'b'], 'type': 'object'}, outputSchema={'properties': {'result': {'type': 'integer'}}, 'required': ['result'], 'type': 'object', 'x-fastmcp-wrap-result': True}, icons=None, annotations=None, meta={'_fastmcp': {'tags': []}}, execution=None)] 结果为：CallToolResult(content=[TextContent(type='text', text='4', annotations=None, meta=None)], structured_content={'result': 4}, meta=None, data=4, is_error=False)

从日志中也可以看出这种方式MCP Server就是MCP Client的一个子进程。

您可以配置其他设置，如环境变量、工作目录或命令参数：

transport=StdioTransport(command="python",args=["my_server.py","--verbose"],env={"LOG_LEVEL":"DEBUG"},cwd="/path/to/server")client=Client(transport)

为了方便，客户端也可以从文件路径推断 STDIO 传输，但这不允许配置：

fromfastmcpimportClient client=Client("my_server.py")# 受限 - 无配置选项

环境变量

由于 STDIO 服务器不会继承您的环境，您需要传递配置的策略。以下是两种常见方法：
选择性转发 仅传递服务器实际需要的变量：

importosfromfastmcp.client.transportsimportStdioTransport required_vars=["API_KEY","DATABASE_URL","REDIS_HOST"]env={var:os.environ[var]forvarinrequired_varsifvarinos.environ}transport=StdioTransport(command="python",args=["server.py"],env=env)client=Client(transport)

从 .env 文件加载 将配置与代码分离：

fromdotenvimportdotenv_valuesfromfastmcp.client.transportsimportStdioTransport env=dotenv_values(".env")transport=StdioTransport(command="python",args=["server.py"],env=env)client=Client(transport)

会话持续

STDIO 传输默认在多个客户端上下文之间维护会话（keep_alive=True）。这通过为多个连接重用同一个子进程来提高性能，但在需要隔离时可以控制。
默认情况下，子进程在连接之间持续存在：

fromfastmcp.client.transportsimportStdioTransport transport=StdioTransport(command="python",args=["server.py"])client=Client(transport)asyncdefefficient_multiple_operations():asyncwithclient:awaitclient.ping()asyncwithclient:# 重用同一个子进程awaitclient.call_tool("process_data",{"file":"data.csv"})

要在连接之间实现完全隔离，请禁用会话持久性：

transport=StdioTransport(command="python",args=["server.py"],keep_alive=False)client=Client(transport)

当您需要完全隔离（例如，在测试套件中）或当服务器状态可能在连接之间引起问题时，请使用 keep_alive=False。

专用 STDIO 传输

FastMCP 提供便利的传输方式，它们是围绕 StdioTransport 的轻量包装器，具有预配置的命令：

• PythonStdioTransport - 对 .py 文件使用 python 命令
• NodeStdioTransport - 对 .js 文件使用 node 命令
• UvStdioTransport - 对 Python 包使用 uv（使用 env_vars 参数）
• UvxStdioTransport - 对 Python 包使用 uvx（使用 env_vars 参数）
• NpxStdioTransport - 对 Node 包使用 npx（使用 env_vars 参数）

对于大多数用例，请直接使用您所需的命令实例化 StdioTransport。这些专用传输主要对客户端推断快捷方式有用。

远程传输

远程传输连接到作为 Web 服务运行的 MCP 服务器。这与 STDIO 传输是根本不同的模式——您的客户端不是启动和管理服务器进程，而是连接到已经运行的、管理自己环境和生命周期的服务。

流式 HTTP 传输

流式 HTTP 是生产部署的推荐传输方式，通过 HTTP 连接提供高效的双向流式传输。

• 类： StreamableHttpTransport
• 服务器兼容性： 使用 mcp run --transport http 运行的 FastMCP 服务器

该传输需要一个 URL，并可选择性地支持用于身份验证和配置的自定义头部：

fromfastmcp.client.transportsimportStreamableHttpTransport# 基本连接transport=StreamableHttpTransport(url="https://api.example.com/mcp")client=Client(transport)# 使用自定义头部进行身份验证transport=StreamableHttpTransport(url="https://api.example.com/mcp",headers={"Authorization":"Bearer your-token-here","X-Custom-Header":"value"})client=Client(transport)

为了方便，FastMCP 还提供身份验证助手：

fromfastmcp.client.authimportBearerAuth client=Client("https://api.example.com/mcp",auth=BearerAuth("your-token-here"))

我们这里就用streamable-http方式启动上面的MCP Server，然后运行下面的客户端代码：

fromfastmcpimportClientfromfastmcp.client.transportsimportStreamableHttpTransport transport=StreamableHttpTransport(url="http://127.0.0.1:8001/mcp")asyncdefmain():asyncwithClient(transport=transport)asclient:# 列出可用工具tools=awaitclient.list_tools()print(f"可用的工具有：{tools}")# 调用add工具result=awaitclient.call_tool("add",{"a":1,"b":3})print(f"结果为：{result}")importasyncio asyncio.run(main())

运行结果：

可用的工具有：[Tool(name='add', title=None, description=':param a: 第一个整数\n:param b: 第二个整数\n:return: 返回两个数字之和', inputSchema={'properties': {'a': {'type': 'integer'}, 'b': {'type': 'integer'}}, 'required': ['a', 'b'], 'type': 'object'}, outputSchema={'properties': {'result': {'type': 'integer'}}, 'required': ['result'], 'type': 'object', 'x-fastmcp-wrap-result': True}, icons=None, annotations=None, meta={'_fastmcp': {'tags': []}}, execution=None)] 结果为：CallToolResult(content=[TextContent(type='text', text='4', annotations=None, meta=None)], structured_content={'result': 4}, meta=None, data=4, is_error=False)

注意：streamable-http协议中客户端访问的端点为mcp，地址中必须以mcp结尾：http://127.0.0.1:8001/mcp。

SSE（遗留）

服务器发送事件传输为向后兼容性而维护，但在新部署中已被流式 HTTP 取代。

• 类： SSETransport
• 服务器兼容性： 使用 mcp run --transport sse 运行的 FastMCP 服务器

SSE 传输支持与流式 HTTP 相同的配置选项：

fromfastmcp.client.transportsimportSSETransport transport=SSETransport(url="https://api.example.com/sse",headers={"Authorization":"Bearer token"})client=Client(transport)

除非您对 SSE 有特定的基础设施要求，否则请在新部署中使用流式 HTTP。
这类也可以用sse传输方式启动MCP Server，用下面的MCP Client进行测试：

fromfastmcpimportClientfromfastmcp.client.transportsimportSSETransport transport=SSETransport(url="http://127.0.0.1:8001/sse")asyncdefmain():asyncwithClient(transport=transport)asclient:# 列出可用工具tools=awaitclient.list_tools()print(f"可用的工具有：{tools}")# 调用add工具result=awaitclient.call_tool("add",{"a":1,"b":3})print(f"结果为：{result}")importasyncio asyncio.run(main())

运行结果：

可用的工具有：[Tool(name='add', title=None, description=':param a: 第一个整数\n:param b: 第二个整数\n:return: 返回两个数字之和', inputSchema={'properties': {'a': {'type': 'integer'}, 'b': {'type': 'integer'}}, 'required': ['a', 'b'], 'type': 'object'}, outputSchema={'properties': {'result': {'type': 'integer'}}, 'required': ['result'], 'type': 'object', 'x-fastmcp-wrap-result': True}, icons=None, annotations=None, meta={'_fastmcp': {'tags': []}}, execution=None)] 结果为：CallToolResult(content=[TextContent(type='text', text='4', annotations=None, meta=None)], structured_content={'result': 4}, meta=None, data=4, is_error=False)

内存传输

内存传输直接连接到同一 Python 进程中的 FastMCP 服务器实例。这消除了子进程管理和网络开销，使其非常适合测试和开发。

• 类： FastMCPTransport

与 STDIO 传输不同，内存服务器可以完全访问您 Python 进程的环境。它们与您的客户端代码共享相同的内存空间和环境变量——不需要隔离或显式环境传递。

fromfastmcpimportFastMCP,Clientimportos mcp=FastMCP("TestServer")@mcp.tooldefgreet(name:str)->str:prefix=os.environ.get("GREETING_PREFIX","Hello")returnf"{prefix},{name}!"client=Client(mcp)asyncwithclient:result=awaitclient.call_tool("greet",{"name":"World"})

MCP JSON 配置传输

此传输支持新兴的 MCP JSON 配置标准，用于定义多个服务器：

• 类： MCPConfigTransport

config={"mcpServers":{"weather":{"url":"https://weather.example.com/mcp","transport":"http"},"assistant":{"command":"python","args":["./assistant.py"],"env":{"LOG_LEVEL":"INFO"}}}}client=Client(config)asyncwithclient:# 工具按服务器命名空间weather=awaitclient.call_tool("weather_get_forecast",{"city":"NYC"})answer=awaitclient.call_tool("assistant_ask",{"question":"What?"})

使用 FastMCP 和 MCPConfig 进行工具转换

FastMCP 支持在 MCPConfig 文件中与 MCP 服务器一起定义基本的工具转换。

config={"mcpServers":{"weather":{"url":"https://weather.example.com/mcp","transport":"http","tools":{}#<---这是工具转换部分}}}

通过这些转换，您可以转换（更改）工具的名称、标题、描述、标签、启用状态和参数。
对于工具采用的每个参数，您可以转换（更改）名称、描述、默认值、可见性、是否必需，并且可以提供示例值。
在以下示例中，我们将 weather_get_forecast 工具转换为仅检索 Miami 的天气，并从客户端隐藏 city 参数。

tool_transformations={"weather_get_forecast":{"name":"miami_weather","description":"获取迈阿密的天气","arguments":{"city":{"name":"city","default":"Miami","hide":True,}}}}config={"mcpServers":{"weather":{"url":"https://weather.example.com/mcp","transport":"http","tools":tool_transformations}}}

工具白名单和黑名单

通过在服务器上为工具应用 tags，可以从客户端将工具列入白名单或黑名单。在以下示例中，我们仅将标记有 forecast 标签的工具列入白名单，所有其他工具对客户端不可用。

tool_transformations={"weather_get_forecast":{"enabled":True,"tags":["forecast"]}}config={"mcpServers":{"weather":{"url":"https://weather.example.com/mcp","transport":"http","tools":tool_transformations,"include_tags":["forecast"]}}}

后记

对于初学者，可以直接关注stdio和streamable-http两种传输方式，可以直接运行的代码如下：

MCP Server

fromfastmcpimportFastMCP# 初始化FastMCP Servermcp=FastMCP("My MCP Server")# 用@mcp.tool()装饰器定义一个工具@mcp.tool()defadd(a:int,b:int)->int:""" :param a: 第一个整数 :param b: 第二个整数 :return: 返回两个数字之和 """returna+b# 启动MCP Serverif__name__=="__main__":# ssemcp.run(transport="sse",host="127.0.0.1",port=8001)# streamable-http# mcp.run(transport="streamable-http", host="127.0.0.1", port=8001)# stdio# mcp.run(transport="stdio")

MCP Client

1. stdio

fromfastmcpimportClientfromfastmcp.client.transportsimportStdioTransport transport=StdioTransport(command="python",args=["my_server.py"])asyncdefmain():asyncwithClient(transport=transport)asclient:# 列出可用工具tools=awaitclient.list_tools()print(f"可用的工具有：{tools}")# 调用add工具result=awaitclient.call_tool("add",{"a":1,"b":3})print(f"结果为：{result}")importasyncio asyncio.run(main())

2. streamable-http

fromfastmcpimportClientfromfastmcp.client.transportsimportStreamableHttpTransport transport=StreamableHttpTransport(url="http://127.0.0.1:8001/mcp")asyncdefmain():asyncwithClient(transport=transport)asclient:# 列出可用工具tools=awaitclient.list_tools()print(f"可用的工具有：{tools}")# 调用add工具result=awaitclient.call_tool("add",{"a":1,"b":3})print(f"结果为：{result}")importasyncio asyncio.run(main())

参考文献

FastMCP客户端传输