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

【最后203篇系列】031 构建MCP尝试

一早就想搭建自己的mcp,但是一直没有合适的契机开始。这个周末正好有点时间看了看,非常推荐B站的几个视频, up主叫「马克的技术工作坊」,我觉得写的非常好,很用心,看了真不会浪费时间。

在这里插入图片描述

所以今天就打算一鼓作气跟着走一遍(可能对于读者的意义在于验证up主内容的实操性,毕竟实操起来是有点差异的)。

从稍微功利一点的角度出发,说说看为什么现在研究mcp。
个人或企业开发MCP(Model Context Protocol)相关服务和应用,主要基于以下目的和商机:

gpt5好像现在的确有点弱智,可能和4o一样,要过阵子才能调整好。所以下面是问deepseek的:


1. 降低AI开发门槛,赋能普通人赚钱

  • 目的:MCP协议让AI从“聊天机器人”升级为“可执行任务的智能助手”,普通人无需编程基础即可指挥AI完成复杂任务,如营销、数据分析、自动化流程等。
  • 商机
    • AI代理服务:个人可搭建MCP服务,如自动生成营销文案、电商运营工具,赚取服务费或订阅收入。
    • 教育培训:教授普通人如何利用MCP赚钱,如开设课程或提供咨询服务。

2. 企业优化运营,提升效率

  • 目的:MCP让企业AI系统能自主规划、执行任务,如广告投放、客户服务、供应链管理等,减少人力成本并提高响应速度。
  • 商机
    • 行业解决方案:如利欧数字的“盘古优化助手”通过MCP优化广告投放,企业可付费使用。
    • 垂直领域AI代理:如工程机械行业的“星知Agent”自动执行营销任务,企业可定制开发类似产品。

3. 构建MCP生态,抢占市场先机

  • 目的:MCP类似于AI时代的“USB-C”接口,标准化AI与外部服务的交互方式。企业可通过提供MCP基础设施(如服务器、应用市场)占据生态位。
  • 商机
    • MCP应用市场:如独立开发者艾逗笔的MCP.so,收录超1万个MCP服务,通过流量变现或API收费。
    • MCP服务器托管:如Kiwi.com与Alpic合作提供航班搜索MCP服务,企业可开发类似行业专用MCP服务器。

4. 推动AI标准化,制定行业规则

  • 目的:MCP协议正在成为AI智能体的通用标准,企业参与标准制定可增强话语权。
  • 商机
    • 行业联盟:如利欧数字联合广告协会制定MCP广告标准,未来可收取认证或授权费用。
    • 合规与安全服务:提供MCP数据隐私、伦理审查等增值服务。

5. 创新商业模式,如AIaaS(AI即服务)

  • 目的:MCP让企业可按需调用AI能力,无需自建大模型,降低使用门槛。
  • 商机
    • 订阅制AI工具:如蝉镜MCP让企业输入一句话即可生成数字人视频,按使用量收费。
    • AI代理租赁:企业可提供特定领域的AI代理(如金融分析、医疗诊断),按任务收费。

总结

  • 个人机会:低门槛AI创业、教育培训、代理服务。
  • 企业机会:行业解决方案、MCP基础设施、标准化服务、AIaaS商业模式。
  • 未来趋势:MCP可能成为AI时代的“操作系统”,早期布局者将占据生态优势。

我觉得机会点在于:

  • 1 基础大模型个人或者说95%的企业都没有能力去开发,所以虽然整体价值大,对大多数人来说不是商机。从23年到25年,基础大模型的竞争只剩下了几个巨头 。本来可能更少的,因为贸易战的原因,现在中美阵营各有大约10个还在竞争。而且这两年的趋势是:大模型本身的能力已经触顶,这个是因为大模型目前的学习方法决定的。这个期间形成了两个重要的基础:①大模型的基础能力可以和人类媲美 ②大模型成本高,使用基本需要付费。顺带的,由于大模型能力触顶,所以能力下溢了。过去的8B模型能达到的水平和现在的8B水平一比较就清除了,所以在今年,可用的本地模型将会出现,这会进一步助力个人/小企业掌握这一基础能力。颇有点当前BERT的架势。
  • 2 基础大模型的能力如果比作电力,那么价值在于怎么用电力来创造产品。Cursor是一个比较有代表性的例子,成功的切入了代码编辑这个领域,可行性高,价值高。所以作为一个初创公司估值极速飙升。伴随着这股潮流,MCP迅速发展,已经有若干个大的MCP市场,整体的逻辑是各个领域全面和大模型对接。
  • 3 所以在现阶段,一个明显的收益点是把当前的业务(对于第三产业),通过MCP和大模型对接。对于这个机会,我的理解是:一方面,有些当前业务的使用非常狭窄,比如番茄酱,单独去卖番茄酱给个人是比较难的,但是很多人如果在吃汉堡或者薯条的时候就想来点,这样就能够卖出去。具体来说,一个用户在用大模型处理一个问题,到了某个关节点突然有个小的需求,就是查一下这个企业是否涉诉。当然,他可以打开一个网页,上企查查之类的去查,但是这样不仅浪费时间,而且不得不把注意力挪开,然后还要把补充的结果和之前的结果融合,很麻烦。这个使用过cursor自动操作服务器开发、调试写文档就有直观感受到了。所以,通过MCP,原来不好卖的东西卖出去了。另一方面,也是之前提到的,用户已经习惯了给大模型付费。想想短信时代,很多公司直接和运营商绑定谈分成:运营商希望用户用更多的短信,如果某些应用能给运营商带来更大的用量,运营商可以把收入的一部分分给这个公司。

更大的机会在于,现在的mcp基本还是群雄混战的年代,有很多技术点还差很远。进一步如果可以实现Agent替代某些工种是更有想象力的机会,有些工作真的完全可以被Agent完美替代。

切入正题,我这里跟着视频里介绍的内容走一遍。

实验环境是我的Macbook,这款笔记本是M1刚出来的时候买的,有比较坑的地方。前一个本子还是intel的芯片,然后M1刚出来的时候为了和前代融合,用了rosetta来运行以前的程序。总体上还是很厉害的,基本上没出什么幺蛾子。但现在越来越多限制了,很多新的东西经常装不了,理论上,应该买一个新的mbp,不要再迁移,而是直接重装。但现在不太行,本子还好好的,再买一个要挨骂…

mbp上还有miniconda,目前基本上在用这个在隔离环境。默认的基础环境是py3.9的,此次搭建至少要py3.10以上的。

  • 1 创建新环境

conda create --name mcp python=3.12

  • 2 安装uv

安装uv,一个和pip类似的包管理器, 也可以用来执行命令。(会比pip快很多)
在这里插入图片描述

方式一:pip安装的方法可能包的版本会偏老

pip3 install uv -i https://mirrors.aliyun.com/pypi/simple/

方式二:采用官方的脚本(我的苹果电脑可能有点问题)

curl -LsSf https://astral.sh/uv/install.sh | sh

装好了之后可以看到版本号还是偏低的,但是对这次实验没有影响

pip3 list |grep uv
uv         0.8.8
  • 3 初始项目

这点真的有点像git …

cd weather

创建虚拟环境

uv venv

激活虚拟环境

source .venv/bin/activate

安装对应的包

uv add “mcp[cli]” httpx

项目文件

README.md      
main.py        
pyproject.toml 
uv.lock       
weather.py

其中weather是加入的具体mcp服务:

其中FastMCP是协议的纽带,其他的类似定一个FastAPI。每个函数的注释很重要,包含了用户和入参,通过这个注释可以自动完成函数的元数据注册。以Cline为例,在完成MCP服务注册后,它会更新自己的system_prompt,把这些信息写到prompt中。Cline的Systemprompt一开始就有接近5万字,快接近deepseekv3的上限了(64k),自动缓存技术在这里还是很友好的,否则成本还得翻几番。

另外,Cline是以XML方式完成交互的,输入和输出是以json schema方式说明的(类似pydantic的作用),这些都是需要了解的地方。

代码先定义了个异步函数,负责从美国气象局查天气:

  • 1 make_nws_request : 根据给到的url进行查询,NWS_API_BASE这个常量决定了接口是查询美国天气局的数据

后面定义了两个异步函数(通过路径拼接,指向了某个城市(point)的数据,或者是某个州的天气报警),通过mcp.tools装饰器,两个异步函数变成了大模型的工具。

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP# Initialize FastMCP server
mcp = FastMCP("weather", log_level="ERROR")# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"async def make_nws_request(url: str) -> dict[str, Any] | None:"""Make a request to the NWS API with proper error handling."""headers = {"User-Agent": USER_AGENT,"Accept": "application/geo+json"}async with httpx.AsyncClient() as client:try:response = await client.get(url, headers=headers, timeout=30.0)response.raise_for_status()return response.json()except Exception:return Nonedef format_alert(feature: dict) -> str:"""Format an alert feature into a readable string."""props = feature["properties"]return f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}
"""@mcp.tool()
async def get_alerts(state: str) -> str:"""Get weather alerts for a US state.Args:state: Two-letter US state code (e.g. CA, NY)"""url = f"{NWS_API_BASE}/alerts/active/area/{state}"data = await make_nws_request(url)if not data or "features" not in data:return "Unable to fetch alerts or no alerts found."if not data["features"]:return "No active alerts for this state."alerts = [format_alert(feature) for feature in data["features"]]return "\n---\n".join(alerts)@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:"""Get weather forecast for a location.Args:latitude: Latitude of the locationlongitude: Longitude of the location"""# First get the forecast grid endpointpoints_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"points_data = await make_nws_request(points_url)if not points_data:return "Unable to fetch forecast data for this location."# Get the forecast URL from the points responseforecast_url = points_data["properties"]["forecast"]forecast_data = await make_nws_request(forecast_url)if not forecast_data:return "Unable to fetch detailed forecast."# Format the periods into a readable forecastperiods = forecast_data["properties"]["periods"]forecasts = []for period in periods[:5]:  # Only show next 5 periodsforecast = f"""
{period['name']}:
Temperature: {period['temperature']}°{period['temperatureUnit']}
Wind: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""forecasts.append(forecast)return "\n---\n".join(forecasts)if __name__ == "__main__":# Initialize and run the servermcp.run(transport='stdio')
  • 4 在Cline中进行MCP服务器配置

原文中的command 是配置 uv ,但是之前说我的mbp的历史,有点混乱,所以最后我是在虚拟环境里找到新的uv路径替换才成功。

conda activate myenv
which uv  # 输出路径如 /path/to/miniconda3/envs/myenv/bin/uv

在这里插入图片描述
在修改配置文件的过程中,Cline会一直扫描配置文件的变动,如果正常,左边就会出现
在这里插入图片描述
展开就会有更多细节
在这里插入图片描述

这时候就可以调用了

在这里插入图片描述
拿到返回的结果,大模型就会开始输出
在这里插入图片描述
因为没有看到mcp服务里怎么查一个城市的经纬度函数,我就又问了一下:原来这里是利用了大模型的常识
在这里插入图片描述
然后我又问了另一个问题,首先大模型先尝试了州的天气预警工具(调用正确)
在这里插入图片描述
发现没有报警,这时候模型又“反思”了一下,决定把Newark的信息查给我
在这里插入图片描述
结果如下:
在这里插入图片描述

这里就要提到up主后面一段的工作了,追查下去,这里有两点可以提:

  • 1 Cline也遵循React的模式,让模型总是在思考标签里进行思考(Think),然后再行动(Act),最后再观察(Observation)。其实这也是强化学习的概念。所以大模型在执行指令的时候考虑“更周到”一些
  • 2 Cline的任务是允许无限执行(Do Until Success)的。当前的状态来说,只能说有好有坏,之前我是遇到到无限死循环的(但是Clien有一个自动执行20轮需要再授权的保护)

总结

到这里,复现先告一段落 。主要还是时间原因,后面up主展开了日志追踪来分析MCP交互的过程,里面有一些方法是高于这个实验本身的,比如做一个代理的OpenAI Compatible服务。

然后我又和大模型确认了一下MCP的定位,在视频中,up主也是提到目前大部分mcp都是走stdio方式通信的。我再推演一点:
应该是没有必要吧,mcp服务如果等同于脚本,起到的是和大模型交互的作用,完成工具注册和调用等功能。而重负载服务可以通过http服务进行,mcp server只要通过异步方式调用就可以了。这样即可以保证本地的轻量,服务商业避免暴露自己的核心逻辑,做到使用和服务分离。

你的设计思路非常合理,尤其是在 轻量化工具调用核心逻辑隔离 方面的权衡。以下是关键点总结和优化建议,进一步验证你的方案:


1. 当前 MCP 服务的定位(工具交互层)

  • 核心作用

    • 工具注册与调用:作为大模型(如 CLINE)的“工具执行器”,按需触发本地脚本或轻量任务。
    • 逻辑隔离:通过 MCP 的配置化调用,隐藏核心实现细节(如脚本路径、参数),仅暴露工具名称和输入输出。
    • 资源友好:无常驻进程,避免资源浪费,适合低频、异步任务。
  • 典型场景

    • 文件处理、数据查询、本地计算等无状态工具
    • 调用外部 HTTP 服务的代理层(如封装第三方 API 的认证逻辑)。

2. 与重负载服务的协作模式

你提到的 “异步调用 HTTP 服务” 是经典的分层架构:

轻量任务
重负载
CLINE 对话
MCP 轻量服务
任务类型
本地脚本/Python
异步调用 HTTP 微服务
  • 优势

    • 安全性:HTTP 服务可部署在内网,通过防火墙/Nginx 限制访问,MCP 仅作为网关转发请求。
    • 解耦:HTTP 服务迭代升级时,MCP 配置无需变更(只要接口协议不变)。
  • 实现示例
    weather.py 中异步调用远程 HTTP 服务:

    import aiohttp
    import asyncioasync def fetch_weather():async with aiohttp.ClientSession() as session:async with session.get("http://internal-weather-api/data") as resp:return await resp.json()# 通过 MCP 调用的入口函数
    def run():return asyncio.run(fetch_weather())
    

3. 如何保持轻量化?

  • 避免轮询/长连接
    MCP 服务每次调用后退出,不维护任何状态(如数据库连接池)。
  • 动态加载依赖
    在脚本内按需导入第三方库(如 import inside_function),减少启动时的内存占用。
  • 超时控制
    你的配置中已有 "timeout": 60,确保异常任务不会卡死。

4. 安全增强建议

  • 最小权限原则
    MCP 的执行用户(如 nobody 或专用账户)仅拥有脚本目录的读/执行权限。
  • 敏感信息隔离
    HTTP 服务的认证信息(如 API Key)不硬编码在脚本中,而是通过环境变量或配置中心获取:
    import os
    API_KEY = os.getenv("WEATHER_API_KEY")
    

5. 性能优化方向

  • 预热脚本
    如果脚本初始化耗时,可通过 uv --preload 提前加载虚拟环境(需测试效果)。
  • 结果缓存
    对高频但数据更新慢的任务(如天气预报),在脚本内添加缓存逻辑(如 cachetools 库)。

总结

你的架构设计(MCP 轻量脚本 + 异步调用重型服务)在以下场景中尤为合适:

  1. 工具化需求:大模型需要灵活调用本地能力时。
  2. 隐私敏感场景:核心逻辑需保留在内部 HTTP 服务中。
  3. 资源受限环境:本地设备(如树莓派)无法常驻多个服务。

如果需要进一步扩展,可以考虑:

  • 为 MCP 添加简单的 权限控制(如调用方白名单)。
  • 使用 消息队列(如 Redis/RabbitMQ)解耦 MCP 和重型服务。

接下来,开发若干实用的MCP服务,进一步掌握这个过程。

对我来说,我可以把自己的Agent封装成MCP服务。原来的MSW设计变为MMcp设计,省去SW的功夫,通过大模型来帮我完成大量不同类型的。

需要注意的是,现在的大模型仍然缺乏可靠的应用方案,所以我需要对M进行限制,一般只暴露查询接口;另外Cline是外部模型,我也会限制其了解我真实的密码等信息,使用资源简称是一个好的方案。

在开发的过程中有一个好处:规范。因为注册模型工具的需要,每个函数必须有完整的注释,这样以后继续进行复用的可能性非常高。

最后,可以借鉴Cline的方式,设计自己的Agent,服务与垂直行业。

http://www.dtcms.com/a/324747.html

相关文章:

  • signed 和 unsigned 类型说明符
  • 嵌套-列表存储字典,字典存储列表,字典存储字典
  • Linux 路由子系统深度分析:框架、实现与代码路径
  • web刷题3
  • 商业解决方案技术栈总结
  • Python 获取对象信息的所有方法
  • 基于Spring Boot和SSE的实时消息推送系统
  • 三数之和 Java
  • 人工智能系列(7)人工神经网络中的无监督学习
  • C语言-数组和指针练习题合集(一)
  • C语言深度剖析
  • 网页五子棋测试
  • VUE+SPRINGBOOT从0-1打造前后端-前后台系统-关于我们
  • 2025最新免费的大模型和免费的大模型API有哪些?(202508更新)
  • 秋招春招实习百度笔试百度管培生笔试题库百度非技术岗笔试|笔试解析和攻略|题库分享
  • 冒泡排序实现以及优化
  • WebSocket集群方案解析与实现
  • My APK 安卓版:高效管理手机应用的工具软件
  • windows的cmd命令【持续更新】
  • Linux应用软件编程---文件操作1(fopen、fclose、fgetc/fputc、fgets/fputs)
  • 什么是浏览器标识?
  • 【Docker进阶实战】从多容器编排到集群部署
  • TSF应用开发与运维部署
  • 个人笔记Mybatis2
  • 医学统计(现况调查的统计分析策略1)
  • 电脑使用“碎片整理”程序的作用
  • 基于ECharts的智慧社区数据可视化
  • 【npm、yarn、pnpm】特点对比,按需选择
  • Java设计模式之开闭原则介绍与说明
  • 【RocketMQ 生产者和消费者】- ConsumeMessageOrderlyService 顺序消费消息