news 2026/4/30 20:54:53

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

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
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())
  1. 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客户端传输

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/4/26 14:23:43

掌握Open-AutoGLM三大调优技巧，快速提升语义解析准确率

第一章&#xff1a;Open-AutoGLM语义解析准确率提升的核心价值提升Open-AutoGLM的语义解析准确率不仅增强了模型对自然语言的理解能力&#xff0c;更在实际应用场景中释放出巨大潜力。高准确率意味着系统能更精确地识别用户意图、还原语义结构&#xff0c;并生成符合上下文逻辑…

作者头像 李华
网站建设 2026/4/30 19:25:54

python实验设备借用平台的 实验室设备租赁系统设计与实现_q68o5k49

文章目录具体实现截图主要技术与实现手段关于我本系统开发思路java类核心代码部分展示结论源码lw获取/同行可拿货,招校园代理 &#xff1a;文章底部获取博主联系方式&#xff01;具体实现截图 同行可拿货,招校园代理 python实验设备借用平台的 实验室设备租赁系统设计与实现…

作者头像 李华
网站建设 2026/5/1 8:04:27

java计算机毕业设计外卖点餐管理系统 基于SpringBoot的在线餐饮订餐平台 Java Web智能外卖配送管理系统

计算机毕业设计外卖点餐管理系统4nd5o9 &#xff08;配套有源码 程序 mysql数据库 论文&#xff09; 本套源码可以在文本联xi,先看具体系统功能演示视频领取&#xff0c;可分享源码参考。 快节奏生活让“外卖”成为刚需&#xff0c;但电话点餐易出错、纸质菜单更新慢、高峰期接…

作者头像 李华
网站建设 2026/4/27 10:02:09

Open-AutoGLM操作序列生成瓶颈：90%开发者忽略的3个致命问题及解决方案

第一章&#xff1a;Open-AutoGLM操作序列生成优化概述Open-AutoGLM 是一种面向自动化任务执行的生成式语言模型框架&#xff0c;专注于优化复杂操作序列的生成过程。该系统通过融合指令理解、上下文建模与动作规划机制&#xff0c;实现对多步骤任务的高效推理与执行路径构建。其…

作者头像 李华
网站建设 2026/5/1 8:00:52

Open-AutoGLM重试机制深度解析（失败处理的黄金法则）

第一章&#xff1a;Open-AutoGLM重试机制概述在构建基于大语言模型&#xff08;LLM&#xff09;的自动化系统时&#xff0c;网络波动、服务限流或模型推理超时等问题可能导致请求失败。Open-AutoGLM 为此设计了健壮的重试机制&#xff0c;以提升任务执行的稳定性和最终成功率。…

作者头像 李华