更新时间:2026-07-17 gmt 08:00
制作mcp server镜像-j9九游会登录
一个最小化的mcp streamable http计算器服务器示例,基于官方mcp sdk的fastmcp实现。适合作为学习mcp协议,快速搭建streamable http服务的起点。
mcp协议介绍
model context protocol (mcp) 是由anthropic提出并开源的协议标准,用于标准化大型语言模型(llm)应用与外部工具、数据源之间的通信方式。它定义了一套统一的json-rpc 2.0消息格式,让ai客户端(如 claude desktop、cursor、claude code 等)能够发现、调用外部工具,而无需关心底层实现细节。
streamable http 核心特性
- 单端点:同一个 url(如 /mcp)处理所有通信,简化路由和负载均衡配置。
- post 请求:客户端向服务器发送 json-rpc 请求消息。
- get 请求(可选):客户端建立 sse 流,用于接收服务器主动推送的消息(如日志、资源变更通知)。
- 内容协商:客户端必须发送 accept: application/json, text/event-stream,服务器根据场景选择返回同步 json 或 sse 流。
- 会话管理:通过 mcp-session-id 头部实现有状态会话,支持断线重连和消息重传。
- 无状态模式:对于简单的工具调用场景,可以直接返回 application/json,无需维护 sse 连接,更适合 serverless 和水平扩展。
制作docker镜像
构建镜像:
#进入mcp代码开发路径 cd mcp_test/ # 生成 arm64 镜像 docker build --platform linux/arm64 -t mcp-calculator:latest .
本地docker调测
本地启动docker容器:
docker run -d \ --name mcp-calculator1 \ -p 8080:8080 \ mcp-calculator:latest
本地调测docker:
# 方式 1:curl 直接调用 initialize
curl -x post http://127.0.0.1:8080/mcp \
-h "content-type: application/json" \
-h "accept: application/json, text/event-stream" \
-d '{
"jsonrpc": "2.0",
"id": 0,
"method": "initialize",
"params": {
"protocolversion": "2025-03-26",
"capabilities": {},
"clientinfo": {"name": "test", "version": "1.0.0"}
}
}'
#调测结果
event: message
data: {"jsonrpc":"2.0","id":0,"result":{"protocolversion":"2025-03-26","capabilities":{"experimental":{},"prompts":{"listchanged":false},"resources":{"subscribe":false,"listchanged":false},"tools":{"listchanged":false}},"serverinfo":{"name":"calculator-mcp-server","version":"1.28.1"}}}
# 方式 2:curl 调用工具
curl -x post http://127.0.0.1:8080/mcp \
-h "content-type: application/json" \
-h "accept: application/json, text/event-stream" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "add",
"arguments": {"a": 3, "b": 5}
}
}'
#调测结果
event: message
data: {"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"8.0"}],"structuredcontent":{"result":8.0},"iserror":false}} 调测完成后停止并删除容器:
docker stop mcp-calculator1 docker rm mcp-calculator1
上传镜像
将构建的镜像上传至容器镜像服务。
- 登录容器镜像服务控制台,选择区域,要和agentarts区域保持一致,否则无法选择到镜像。
- 单击右上角“创建组织”,输入组织名称完成组织创建。请自定义组织名称,本示例使用“demo-886633”,下面的命令中涉及到组织名称“demo”也请替换为自定义的值。
- 单击右上角“登录指令”,获取登录访问指令,本文选择复制“通用型登录指令”登录指令。
以root用户登录本地环境,输入复制的swr临时登录指令。
- 上传镜像至容器镜像服务镜像仓库。
- 使用docker tag命令给上传镜像打标签。
#region和domain信息请替换为实际值,组织名称deep-learning也请替换为自定义的值。 sudo docker tag mcp-calculator:latest swr.{region-id}.{domain}/demo-886633/mcp-calculator:latest #此处以华为云cn-southwest-2为例 sudo docker tag mcp-calculator:latest swr.cn-southwest-2.myhuaweicloud.com/demo-886633/mcp-calculator:latest - 使用docker push命令上传镜像。
#region和domain信息请替换为实际值,组织名称deep-learning也请替换为自定义的值。 sudo docker push swr.{region-id}.{domain}/demo-886633/mcp-calculator:latest #此处以华为云cn-southwest-2为例 sudo docker push swr.cn-southwest-2.myhuaweicloud.com/demo-886633/mcp-calculator:latest
- 使用docker tag命令给上传镜像打标签。
- 完成镜像上传后,在容器镜像服务控制台的“我的镜像”页面可查看已上传的自定义镜像。
部署到华为云agentarts
- 部署智能体运行时,部署时“入栈协议”请选择为“mcp”,具体请参考通过控制台部署智能体运行时。 图1 部署智能体运行时
- 云上agentarts调测。
- 列出工具
curl -x post https://${runtime_endpoint} \ # runtime_endpoint的获取方式:在智能体运行时列表单击名称,在详情页面的"访问方式"中的"访问地址" -h "content-type: application/json" \ -h 'x-hw-agentarts-session-id: {xxxxxxxxxx}' -h 'authorization: bearer {apikey}' -h "accept: application/json, text/event-stream" \ -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' - 调用工具
curl -x post https://${runtime_endpoint} \ -h "content-type: application/json" \ -h "accept: application/json, text/event-stream" \ -h 'authorization: bearer {apikey}' -h "accept: application/json, text/event-stream" \ -d '{ "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "add", "arguments": {"a": 3, "b": 5} } }'如果返回了正确的 json-rpc 结果,说明云上部署和调测成功。
- 列出工具
代码开发示例
项目结构
- 创建项目文件夹:
mkdir mcp_test
- 项目下的工程目录文件:
mcp_test/ ├── mcp_server.py # 主服务器实现 ├── requirements.txt # python 依赖 ├── dockerfile # 自定义镜像构建
完整服务器代码
mcp_server.py:
import logging
import os
import uvicorn
from mcp.server import transport_security
from mcp.server.fastmcp import fastmcp
from mcp.server.streamable_http_manager import transportsecuritysettings
from starlette.requests import request
from starlette.responses import jsonresponse
# 配置日志
logging.basicconfig(level=logging.info)
logger = logging.getlogger(__name__)
transport_security.transportsecuritymiddleware._validate_host = lambda _self, _host: true
logger.info("mcp host header validation patched to allow all hosts")
# 创建 mcp 服务器实例
mcp = fastmcp(
"calculator-mcp-server",
streamable_http_path="/mcp",
stateless_http=true,
transport_security=transportsecuritysettings(
enable_dns_rebinding_protection=false,
),
)
logger.info(f"mcp transport_security settings: {mcp.settings.transport_security}")
@mcp.tool()
def add(a: float, b: float) -> float:
"""两个数字相加"""
logger.info(f"add called: {a} {b}")
return a b
@mcp.tool()
def subtract(a: float, b: float) -> float:
"""两个数字相减"""
logger.info(f"subtract called: {a} - {b}")
return a - b
@mcp.tool()
def multiply(a: float, b: float) -> float:
"""两个数字相乘"""
logger.info(f"multiply called: {a} * {b}")
return a * b
@mcp.tool()
def divide(a: float, b: float) -> float:
"""两个数字相除,除数不能为0"""
logger.info(f"divide called: {a} / {b}")
if b == 0:
raise valueerror("除数不能为0")
return a / b
@mcp.custom_route("/ping", methods=["get"])
async def ping(_request: request) -> jsonresponse:
"""网关探活 ping 接口"""
return jsonresponse({"status": "ok", "message": "pong"})
# 导出 streamable http asgi 应用
app = mcp.streamable_http_app()
def main() -> none:
"""命令行入口,供 uvicorn / 控制台脚本调用。"""
port = int(os.getenv("port", 8080))
uvicorn.run(app, host="0.0.0.0", port=port)
if __name__ == "__main__":
main() dockerfile:
# 先配置docker内置镜像加速拉取基础镜像,构建阶段pip用华为云源
from python:3.11-slim
workdir /app
copy requirements.txt .
# 华为云pip临时源
run pip install --no-cache-dir --upgrade pip \
-i https://mirrors.huaweicloud.com/repository/pypi/simple \
--trusted-host mirrors.huaweicloud.com \
&& pip install --no-cache-dir -r requirements.txt \
-i https://mirrors.huaweicloud.com/repository/pypi/simple \
--trusted-host mirrors.huaweicloud.com
copy mcp_server.py .
expose 8080
cmd ["uvicorn", "mcp_server:app", "--host", "0.0.0.0", "--port", "8080", "--proxy-headers", "--forwarded-allow-ips", "*"] requirements.txt:
mcp==1.28.1 uvicorn==0.46.0
本地验证
参考环境准备做好环境准备工作。
# 1. 安装依赖
pip install -r requirements.txt -i https://mirrors.huaweicloud.com/repository/pypi/simple --trusted-host mirrors.huaweicloud.com
# 2. 启动服务器
python mcp_server.py
# 3. 测试 initialize
curl -x post http://127.0.0.1:8080/mcp \
-h "content-type: application/json" \
-h "accept: application/json, text/event-stream" \
-d '{"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolversion":"2025-03-26","capabilities":{},"clientinfo":{"name":"test","version":"1.0.0"}}}'
# 测试结果
event: message
data: {"jsonrpc":"2.0","id":0,"result":{"protocolversion":"2025-03-26","capabilities":{"experimental":{},"prompts":{"listchanged":false},"resources":{"subscribe":false,"listchanged":false},"tools":{"listchanged":false}},"serverinfo":{"name":"calculator-mcp-server","version":"1.28.1"}}}
# 4. 测试工具调用
curl -x post http://127.0.0.1:8080/mcp \
-h "content-type: application/json" \
-h "accept: application/json, text/event-stream" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"add","arguments":{"a":3,"b":5}}}'
# 测试结果
event: message
data: {"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"8.0"}],"structuredcontent":{"result":8.0},"iserror":false}}
#停止服务器
ps -ef |grep mcp_server.py
root 12692 5936 0 15:29 pts/0 00:00:00 python mcp_server.py
root 13674 5936 0 15:37 pts/0 00:00:00 grep --color=auto mcp_server.py
(venv) root@ecs-tmp-ggq:~/mcp_test# kill -9 12692 父主题:
相关文档
意见反馈
文档内容是否对您有帮助?
提交成功!非常感谢您的反馈,我们会继续努力做到更好! 您可在查看反馈及问题处理状态。
系统繁忙,请稍后重试
如您有其它疑问,您也可以通过华为云社区问答频道来与我们联系探讨