当前位置： 首页 > news >正文

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

news 2026/7/23 2:17:30

前言

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客户端传输

http://www.cnnetsun.cn/news/166867.html

相关文章：

  • Open-AutoGLM操作序列优化实战（专家级调优技巧大公开）
  • 【大模型系统容错进阶指南】：Open-AutoGLM重试机制的7大设计模式
  • Open-AutoGLM坐标偏差高达2米？教你4种高效校正方法
  • 语义解析准确率提升关键路径，Open-AutoGLM最新迭代深度解读
  • 收藏！一文读懂RAG技术核心（附大模型从入门到实战全套学习礼包）
  • 重磅消息！ESXi 8.0 系列推出ESXi 8.0 Update 3h 更新重要版本啦
  • PFC5.0柔性/刚性纤维三点弯曲参数可自定义
  • 基于python文化旅游信息公开管理平台的设计与实现_5257y2x6
  • 【保姆级教程】Attention机制全解析！用PyTorch手写Transformer，大模型开发入门到精通！
  • 前端table表格，零基础入门到精通，收藏这篇就够了
  • 如何将Open-AutoGLM操作延迟降低85%？资深架构师亲授调优心法
  • 从入门到精通：掌握Open-AutoGLM推理优化的7个关键步骤
  • 【Open-AutoGLM macOS适配终极指南】：手把手教你完成全流程配置与优化
  • 【Open-AutoGLM语义解析突破】：准确率提升35%背后的核心技术揭秘
  • 为什么顶尖团队都在用Open-AutoGLM做高效特征提取？真相曝光
  • 【AI模型预处理新纪元】：Open-AutoGLM如何实现毫秒级特征输出
  • 9 个降AI率工具，专科生必备避坑指南
  • 基于数学模型的疫苗接种策略分析
  • 基于时空图神经网络的交通流量预测方法研究
  • 【Open-AutoGLM GPU加速适配全攻略】：手把手教你7步完成高性能推理优化
  • Open-AutoGLM特征提取提速5倍：背后你不知道的3大关键技术
  • [特殊字符]工业标准文档“消化不良“？LLM+知识图谱三步翻倍表格任务F1，钢铁直男秒变逻辑大师！
  • RAG技术揭秘：如何通过检索增强生成解决大模型知识过时与幻觉问题？
  • 【国产大模型端侧落地新突破】：Open-AutoGLM推理效率提升实战
  • Open-AutoGLM端侧部署性能调优，手把手教你榨干NPU算力
  • 网络安全5大子方向！哪个才是最优选择？
  • macOS m芯片配置python低版本失败解决方案
  • 为什么顶尖工程师都在用Open-AutoGLM做任务提醒？真相曝光
  • 【Open-AutoGLM高分通过率背后的真相】：顶尖学员不愿公开的6个刷题技巧
  • 白话大模型与知识库的基础原理（不费脑版）