【MCP系列教程】 Python 实现 FastMCP StreamableHTTP MCP:在通义灵码 IDE 开发并部署至阿里云百炼

本文涉及的产品
多模态交互后付费免费试用,全链路、全Agent
简介: 本文介绍了如何利用 FastMCP 实现 StreamableHTTP MCP 的开发、测试、部署与使用,全面展示全链路实践过程。

image.png

在大模型与智能体技术迅猛发展的背景下,模型与工具之间的实时通信能力正变得愈发关键。

本文将带你从零开始,实战演练如何通过 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当中创建了两个工具imageSynthesistasks ,其中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)进行交流反馈!

1bb53390962d38c328f19fcc8419d77a.png

相关文章
|
人工智能 运维 数据可视化
阿里云百炼 MCP服务使用教程合集
阿里云百炼推出首个全生命周期MCP服务,支持一键部署、无需运维,具备高可用与低成本特点。该服务提供多类型供给、低成本托管及全链路工具兼容,帮助企业快速构建专属智能体。MCP(模型上下文协议)作为标准化开源协议,助力大模型与外部工具高效交互。教程涵盖简单部署、GitHub运营、数据分析可视化及文档自动化等场景,助您快速上手。欢迎加入阿里云百炼生态,共同推动AI技术发展!
5844 0
|
20天前
|
缓存 自然语言处理 API
阿里云百炼产品月刊【2025年8月】
阿里云百炼平台8月推出多项更新与活动。通义千问系列重磅升级,新增多款图像、语音及研究模型,如Qwen-Image、Qwen-Image-Edit、Qwen-MT-Image、Wan2.2-S2V等,全面增强图文生成与编辑能力。推出Qwen-Flash轻量模型,优化代码与推理性能,支持高并发低延迟场景。平台服务稳定性提升,部分模型计费策略调整,上下文缓存价格降低至input_token的20%,并提供100万免费token额度。同步上线“实训Agent创客”活动,助力用户快速上手新模型,提升实践能力。
201 5
|
30天前
|
机器学习/深度学习 人工智能 算法
AI 基础知识从 0.6 到 0.7—— 彻底拆解深度神经网络训练的五大核心步骤
本文以一个经典的PyTorch手写数字识别代码示例为引子,深入剖析了简洁代码背后隐藏的深度神经网络(DNN)训练全过程。
458 56
|
5月前
|
人工智能 网络协议 Linux
MCP 协议: Streamable HTTP 是最佳选择
随着AI应用变得越来越复杂并被广泛部署,原有的通信机制面临着一系列挑战。近期MCP仓库的PR #206引入了一个全新的Streamable HTTP传输层替代原有的HTTP+SSE传输层。本文将详细分析该协议的技术细节和实际优势。
2944 100
|
12天前
|
机器学习/深度学习 人工智能 自然语言处理
B站开源IndexTTS2,用极致表现力颠覆听觉体验
在语音合成技术不断演进的背景下,早期版本的IndexTTS虽然在多场景应用中展现出良好的表现,但在情感表达的细腻度与时长控制的精准性方面仍存在提升空间。为了解决这些问题,并进一步推动零样本语音合成在实际场景中的落地能力,B站语音团队对模型架构与训练策略进行了深度优化,推出了全新一代语音合成模型——IndexTTS2 。
811 23
|
27天前
|
存储 消息中间件 人工智能
Fluss:重新定义实时数据分析与 AI 时代的流式存储
Apache Fluss(孵化中)是新一代流式存储系统,旨在解决传统架构中数据重复复制、高成本与复杂性等问题。它基于 Apache Arrow 构建,支持列式存储、实时更新与高效查询,融合流处理与湖仓架构优势,适用于实时分析、AI 与多模态数据场景。Fluss 提供统一读写、冷热分层与开放生态,已在阿里巴巴大规模落地,助力企业实现低成本、高效率的实时数据处理。
218 25
|
23天前
|
人工智能 Kubernetes 监控
初探:从0开始的AI-Agent开发踩坑实录
本文主要阐述作者通过亲身实践,探索利用AI Agent实现开源应用Helm Chart自动化生成的实践历程。
283 17
初探:从0开始的AI-Agent开发踩坑实录
|
20天前
|
人工智能 编解码 数据可视化
AI创作更自由: 魔搭FLowBench云端工作流上线AIGC专区!支持QwenImageEdit免费出图!
很高兴向大家宣布,ModelScope AIGC 专区的工作流功能正式上线!
296 22