在大模型与智能体技术迅猛发展的背景下，模型与工具之间的实时通信能力正变得愈发关键。
本文将带你从零开始，实战演练如何通过 FastMCP 实现 StreamableHTTP MCP 的开发、测试、部署与使用，完整呈现全链路的案例实践过程。
阿里云百炼平台近期新增了对 StreamableHTTP MCP 的支持，进一步提升了模型与外部系统交互的效率与灵活性。
那么，我们为何要选择 StreamableHTTP MCP？相较于 SSE（Server-Sent Events）和 Stdio（标准输入输出），它又具备哪些优势？
场景 |
StreamableHTTP |
SSE |
Stdio |
远程实时通信 |
✅ 支持双向通信、流式传输和断点续传 |
❌ 仅单向通信，需额外维护连接 |
❌ |
云原生/微服务 |
✅ 无状态设计，适合 Serverless 和水平扩展 |
❌ 长连接资源消耗高 |
❌ |
企业级部署 |
✅ 兼容防火墙、代理和负载均衡器 |
❌ 长连接可能被强制终止 |
❌ |
AI 代理/大模型交互 |
✅ 支持多轮对话、流式反馈和上下文维护 |
❌ 无法维护上下文 |
❌ |
学习 StreamableHTTP 的核心价值在于：
- 简化架构：单一端点 + 动态升级机制降低开发复杂度。
- 提升可靠性：断点续传 + 无状态设计增强系统健壮性。
- 适应现代需求：支持双向通信、云原生部署和企业级兼容性。
对于需要实时通信的场景（如 AI 代理、物联网、实时数据推送），StreamableHTTP 是比 SSE 和 Stdio 更优的选择，尤其适合追求扩展性和未来兼容性的开发者。
其他MCP相关教程：
阿里云百炼 MCP服务使用教程合集：https://developer.aliyun.com/article/1661254
整体思路
使用 Python 搭建一个基于 StreamableHTTP 的 MCP 服务器，对通义万象的 qwen-image 文生图模块中的两个 API 接口进行封装，将其转换为 MCP 服务。
随后，我们将该服务部署至阿里云函数计算，并最终集成至阿里云百炼智能体平台的 MCP 体系中。
下面将介绍具体详细步骤。
一、使用 FastMCP 创建基于 StreamableHTTP 的 MCP 服务，结合通义灵码实现 MCP 功能快速测试与应用
FastMCP官网：https://gofastmcp.com/
FastMCP是什么？
FastMCP 2.0 是构建 Model Context Protocol（MCP）服务器与客户端的高效 Python 框架，被誉为“AI 的 USB-C 接口”。
它让开发者能以简洁、Pythonic 的方式为大语言模型暴露数据（Resources）、功能（Tools）和交互模板（Prompts）。
通过极简的装饰器语法，几行代码即可创建 MCP 服务，大幅降低协议实现复杂度。
相比 1.0 版本（已并入官方 MCP SDK），2.0 提供完整生态支持，涵盖客户端、认证、部署、测试及生产级集成，助力快速构建和发布 AI 工具。由 Prefect 团队打造，是迈向 MCP 生产实践的理想选择。
1、环境准备与项目搭建
1.1 命令行当中安装uv
什么是uv：uv是一款使用Rust语言构建的高性能Python包管理工具，目标是取代传统的工具链（例如pip、venv、pip-tools等）。
在使用前，请确保您的电脑已安装Python，这里建议选择Python 3.10版本进行安装。
pip install uv
如果uv安装失败请参考官方文档：https://docs.astral.sh/uv/
1.2 项目搭建
在您的电脑上创建并打开任意一个文件夹。
如果您使用的是 Windows 系统，可以在文件夹路径的地址栏中直接输入 cmd，即可快速打开命令行窗口。
1）创建Python项目文件
uv init --package --python 3.10 bailian-streamablehttp-mcp-server
请注意，上述操作要求 Python 版本与此处的保持一致，否则在阿里云百炼平台上可能会导致错误。
cd bailian-streamablehttp-mcp-server
2）创建虚拟环境并安装所需的包
在命令行中，使用“uv venv”命令创建虚拟环境，然后激活对应的虚拟环境，并安装所需的依赖项。
注意这里使用的fastmcp的版本为 "2.11.3"。
# 创建虚拟环境 uv venv # 在MacOs/Linux当中激活虚拟环境 source .venv/bin/activate # 在Windows当中激活虚拟环境 .venv\Scripts\activate # 安装依赖库 uv add fastmcp==2.11.3
1.3 在PyCharm中打开刚刚命令行创建的文件夹
打开后，您会看到左侧的文件结构已准备完毕。
备注：如果未安装 PyCharm，也可以使用 VSCode 打开。
1.4 实现 StreamableHttp MCP Server 代码
在src/bailian-streamablehttp-mcp-server下面创建server.py文件，用于搭建StreamableHttp MCP Server服务。
from fastmcp import FastMCP from fastmcp.server.dependencies import get_http_request import requests from starlette.requests import Request mcp = FastMCP("text2image") # 请求URL image_synthesis_url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis" tasks_url = "https://dashscope.aliyuncs.com/api/v1/tasks/" @mcp.tool(name='通义万象qwen-image文生图，生成图片创建任务获取任务ID（发送一个请求创建任务，该请求会返回任务ID（task_id））', description='prompt：输入的提示词（支持中英文，长度不超过800个字符，每个汉字/字母占一个字符，超过部分会自动截断），' 'size：图片大小（输出图像的分辨率，格式为宽*高。默认分辨率为1328*1328。可选的分辨率及其对应的图像宽高比例为：1664*928：16:9。1472*1140：4:3 。1328*1328（默认值）：1:1。1140*1472：3:4。928*1664：9:16），' 'prompt_extend：是否使用提示词扩展（是否开启prompt智能改写。开启后会使用大模型对输入prompt进行智能改写，仅对正向提示词有效。对于较短的输入prompt生成效果提升明显，但会增加3-4秒耗时。true：默认值，开启智能改写。false：不开启智能改写。），' 'watermark：是否使用水印（是否添加水印标识，水印位于图片右下角，文案为“Qwen-Image生成”。false：默认值，不添加水印。true：添加水印。），' ) def imageSynthesis(prompt: str,size:str,prompt_extend:bool,watermark:bool) -> any: request: Request = get_http_request() api_key = request.headers.get("api_key") # 请求头 headers = { 'X-DashScope-Async': 'enable', 'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json' } if not api_key: raise ValueError("请设置 请求头 api_key") # 请求数据 data = { "model": 'qwen-image', "input": { "prompt": prompt }, "parameters": { "size": size, "n": 1, "prompt_extend": prompt_extend, "watermark": watermark } } # 发起POST请求 response = requests.post( url=image_synthesis_url, headers=headers, json=data, timeout=30 ) # 检查响应状态 response.raise_for_status() # 解析响应数据 result = response.json() return result @mcp.tool(name='通义万象qwen-image文生图，根据任务ID查询结果）', description='task_id 任务ID。' ) def tasks(tasks_id: str) -> any: if not tasks_id: raise ValueError("task_id 任务ID不能为空") get_tasks_url = tasks_url + tasks_id request: Request = get_http_request() api_key = request.headers.get("api_key") if not api_key: raise ValueError("请设置 请求头 api_key") # 请求头 headers = { "Authorization": f"Bearer {api_key}" } # 发起GET请求 response = requests.get( get_tasks_url, headers=headers, timeout=30 ) # 检查响应状态 response.raise_for_status() # 解析响应数据 result = response.json() return result def run(): # Start an HTTP server on port 8000 mcp.run(transport="http", host="127.0.0.1", port=8000)
1.5 在__init__.py当中引入上述的server.py
from server import run if __name__ == "__main__": run()
1.6 介绍一下上述代码的核心
在server.py当中创建了两个工具imageSynthesis和tasks ，其中imageSynthesis用于生成图片创建任务获取任务ID，tasks 通过上一个工具生成的tasks_id查询对应的任务进度，并得到对应的图片。
其中 mcp.run(transport="http", host="127.0.0.1", port=8000)定义了当前MCP服务是StreamableHTTP。
如果想切换为SSE或者Stdio，具体可以参考FastMCP的官方文档：https://gofastmcp.com/
2、在通义灵码中对前述搭建的 StreamableHTTP MCP 服务进行测试与验证
运行__init__.py
运行成功。
在 PyCharm 中配置通义灵码插件，设置并集成对应的 MCP 服务。
如果不会安装通义灵码请参考：https://help.aliyun.com/zh/lingma
填写如下信息：
名称：text2image
类型：SSE 或 Streamable HTTP
服务地址：http://127.0.0.1:8000/mcp
api_key：阿里云百炼的密钥可以到阿里云百炼控制台申请https://bailian.console.aliyun.com/
填写完上述信息以后，点击立即添加。
添加成功。
回答对话界面，测试一下
输入提示词：生成一张关于，人工智能的海报图，内容是 全链路大模型服务与应用开发平台。
生成成功
生成成功。
二、将服务部署到阿里云函数计算当中
为什么要将服务部署到函数计算？
我们此前搭建的服务是一个本地服务，运行在本地环境中，无法直接被在线的智能体或大模型所访问和使用。为了实现服务的远程调用与集成，我们需要将其部署到具备公网访问能力的云环境中。
函数计算（FC）作为一种事件驱动的无服务器计算服务，非常适合用于部署此类轻量级服务。
通过将服务上传至函数计算，我们可以获得一个可供外部访问的公网地址，从而使得部署的服务能够在阿里云百炼等在线平台中被调用和使用。
最终，借助该公网地址，我们可以在阿里云百炼平台中顺利接入并使用上述 MCP 服务，实现更广泛的应用与测试场景。
1、打开阿里云函数计算
函数官网：https://www.aliyun.com/product/fc
点击管理控制台
2、创建FastMCP的层
由于 FastMCP 具有较多的项目依赖，为了更好地管理和部署这些依赖项，我们需要创建一个函数计算层来集中存放和配置相关依赖包。
通过使用层，不仅可以有效减小函数部署包的体积，还能提升依赖管理的灵活性和复用性，从而更高效地完成项目的部署与运行。
将层的名称设置为 fastmcp，同时将依赖文件 requirements.txt 中的内容也指定为 fastmcp==2.11.3。
这样做可以确保函数计算层与所需依赖包的名称保持一致，便于识别和管理，同时也保证了在安装依赖时能够正确获取和安装对应的库版本。
其余配置和下图一致。
点击创建
创建成功
3、创建函数，选择Web函数
由于 StreamableHTTP 实现的 MCP 本质上是一个 Web 服务，因此我们在函数计算中选择创建一个“Web 函数”类型。
这种函数类型专为处理 HTTP 请求设计，能够更好地支持和适配 Web 服务的运行需求，确保 MCP 服务在云端可以被正常访问和调用。
创建一个名为 textToimage 的函数，并按照下图所示完成其余的配置项填写。
请根据界面提示依次设置运行环境、执行角色、内存规格、超时时间等相关参数，确保函数能够正常运行并与其他服务顺利集成。
点击创建
函数创建成功后，我们需要在其对应的配置中引用刚刚创建好的层。
首先，进入函数的配置管理界面，找到“高级配置”部分，然后点击“编辑”按钮，准备对函数的依赖层进行设置。
删除Flask，添加自定义的fastmcp层。
设置好以后点击部署。
4、引入之前写的代码，实现功能并稍微对代码进行修改
返回到函数计算的代码编辑界面，在app.py同级目录下创建一个名为 server.py 的文件。
需要注意的是，本次设置的端口号为 9000，这是因为在创建函数时已指定该端口；
如果希望使用 8000 端口，建议在函数创建时进行相应设置。
在代码中，将 host 参数配置为 "0.0.0.0"，以确保服务能够监听所有网络接口。
全部代码如下
from fastmcp import FastMCP from fastmcp.server.dependencies import get_http_request import requests from starlette.requests import Request mcp = FastMCP("text2image") # 请求URL image_synthesis_url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text2image/image-synthesis" tasks_url = "https://dashscope.aliyuncs.com/api/v1/tasks/" @mcp.tool(name='通义万象qwen-image文生图，生成图片创建任务获取任务ID（发送一个请求创建任务，该请求会返回任务ID（task_id））', description='prompt：输入的提示词（支持中英文，长度不超过800个字符，每个汉字/字母占一个字符，超过部分会自动截断），' 'size：图片大小（输出图像的分辨率，格式为宽*高。默认分辨率为1328*1328。可选的分辨率及其对应的图像宽高比例为：1664*928：16:9。1472*1140：4:3 。1328*1328（默认值）：1:1。1140*1472：3:4。928*1664：9:16），' 'prompt_extend：是否使用提示词扩展（是否开启prompt智能改写。开启后会使用大模型对输入prompt进行智能改写，仅对正向提示词有效。对于较短的输入prompt生成效果提升明显，但会增加3-4秒耗时。true：默认值，开启智能改写。false：不开启智能改写。），' 'watermark：是否使用水印（是否添加水印标识，水印位于图片右下角，文案为“Qwen-Image生成”。false：默认值，不添加水印。true：添加水印。），' ) def imageSynthesis(prompt: str,size:str,prompt_extend:bool,watermark:bool) -> any: request: Request = get_http_request() api_key = request.headers.get("api_key") # 请求头 headers = { 'X-DashScope-Async': 'enable', 'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json' } if not api_key: raise ValueError("请设置 请求头 api_key") # 请求数据 data = { "model": 'qwen-image', "input": { "prompt": prompt }, "parameters": { "size": size, "n": 1, "prompt_extend": prompt_extend, "watermark": watermark } } # 发起POST请求 response = requests.post( url=image_synthesis_url, headers=headers, json=data, timeout=30 ) # 检查响应状态 response.raise_for_status() # 解析响应数据 result = response.json() return result @mcp.tool(name='通义万象qwen-image文生图，根据任务ID查询结果）', description='task_id 任务ID。' ) def tasks(tasks_id: str) -> any: if not tasks_id: raise ValueError("task_id 任务ID不能为空") get_tasks_url = tasks_url + tasks_id request: Request = get_http_request() api_key = request.headers.get("api_key") if not api_key: raise ValueError("请设置 请求头 api_key") # 请求头 headers = { "Authorization": f"Bearer {api_key}" } # 发起GET请求 response = requests.get( get_tasks_url, headers=headers, timeout=30 ) # 检查响应状态 response.raise_for_status() # 解析响应数据 result = response.json() return result def run(): # Start an HTTP server on port 9000 mcp.run(transport="http", host="0.0.0.0", port=9000)
在 app.py 中完善相关代码，引入此前创建的 server.py 文件。
from server import run if __name__ == "__main__": run()
5、部署函数
部署代码。
部署代码完成后，进入触发器管理界面，找到对应的触发器并点击“编辑”。
在编辑页面中，将“认证方式”设置为“无需认证”，以取消对该函数调用的身份验证要求，使函数可通过公开访问的方式被调用。
点击确定。
三、将对应部署到函数计算的MCP服务，部署到阿里云百炼当中
复制上述函数计算当中的公网地址
打开阿里云百炼。创建自定义MCP
https://bailian.console.aliyun.com/?tab=mcp#/mcp-market
配置如下：
url：为函数计算的公网地址/mcp
api_key：为阿里云百炼的密钥
{ "mcpServers": { "text2image": { "url": "https://xxxxxxx-hangzhou.xxxx.run/mcp", "type": "streamableHttp", "headers": { "api_key": "sk-1xxxxxxxxxxxxxxxxxx3xx" } } } }
url：为函数计算公网地址
type：设置当前MCP的模式为streamableHttp
headers：设置请求头信息
api_key：阿里云百炼大模型的密钥
然后点击提交部署
回到MCP管理页面
看到对应的MCP已经部署成功
工具显示也没有问题
四、在阿里云百炼智能体当中使用搭建好的MCP服务
创建智能体，使用这个自定义MCP。
选择好对应的模型以后，选择自定义MCP当中text2image。
输入如下提示词：生成一张关于，人工智能的海报图，内容是 全链路大模型服务与应用开发平台
然后查询相关内容
运行成功
🌴注意注意：
可直接点击阿里云百炼 MCP服务使用教程合集链接跳转回合集文章页面。
可直接点击下面链接直接进控制台创建：
👉阿里云百炼详情了解可点击此官网链接：阿里云百炼官网介绍
👉阿里云百炼控制台页面可点击此链接直接进入：阿里云百炼控制台
如果在过程中有任何的疑问都可以在评论区中留言探讨或是加入我们的官方支持群（群号：120480015429）进行交流反馈！
