更新时间: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

上传镜像

将构建的镜像上传至容器镜像服务。

  1. 登录容器镜像服务控制台,选择区域,要和agentarts区域保持一致,否则无法选择到镜像。
  2. 单击右上角“创建组织”,输入组织名称完成组织创建。请自定义组织名称,本示例使用“demo-886633”,下面的命令中涉及到组织名称“demo”也请替换为自定义的值。
  3. 单击右上角“登录指令”,获取登录访问指令,本文选择复制“通用型登录指令”登录指令。

    以root用户登录本地环境,输入复制的swr临时登录指令。

  4. 上传镜像至容器镜像服务镜像仓库。
    1. 使用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
    2. 使用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
  5. 完成镜像上传后,在容器镜像服务控制台的“我的镜像”页面可查看已上传的自定义镜像。

部署到华为云agentarts

  1. 部署智能体运行时,部署时“入栈协议”请选择为“mcp”,具体请参考通过控制台部署智能体运行时
    图1 部署智能体运行时
  2. 云上agentarts调测。
    1. 列出工具
      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"}'
    2. 调用工具
      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 结果,说明云上部署和调测成功。

代码开发示例

项目结构

  1. 创建项目文件夹:
    mkdir mcp_test
  1. 项目下的工程目录文件:
    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

相关文档

网站地图