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

【MCP】从0到1实现一个MCP Server

说好了一辈子,少一年、一个月、一天、一个时辰,都不是一辈子。

《霸王别姬》(1993)

01 MCP介绍

MCP全称Model Context Protocol,模型上下文协议。

MCP 是一种开放协议,它标准化了应用程序如何为 LLM 提供上下文。将 MCP 想象成 AI 应用程序的 USB-C 端口。正如 USB-C 提供了一种将设备连接到各种外围设备和配件的标准化方式一样,MCP 也提供了一种将 AI 模型连接到不同数据源和工具的标准化方式。

02 为什么选择MCP

MCP 可帮助您在 LLM 之上构建代理和复杂的工作流程。LLM 经常需要与数据和工具集成,而 MCP 提供:

  • 越来越多的预构件集成,您的 LLM 可以直接引入并使用
  • 在 LLM 提供商和供应商之间切换的灵活性
  • 在基础架构中保护数据的最佳实践

03 MCP架构

MCP 的核心遵循客户端-服务器架构,其中主机应用程序可以连接到多个服务器:

在这里插入图片描述

  • MCP Hosts:希望通过 MCP 访问数据的 Claude Desktop、IDE 或 AI 工具等程序
  • MCP Clients:与服务器保持 1:1 连接的协议客户端
  • MCP Servers:轻量级程序,每个程序都通过标准化的 Model Context Protocol 公开特定功能
  • Local Data Sources:MCP 服务器可以安全访问的计算机文件、数据库和服务
  • Remote Services:MCP 服务器可以连接到的 Internet 上可用的外部系统(例如,通过 API)

04 快速实现简单的MCP Server

这里我们将快速构建一个简单的MCP服务,以便在 Claude for Desktop 和其他客户端(Cline、Cursor等等)中使用。

MCP 服务可以提供三种主要类型的功能:

  • Resources:客户端可以读取的类似文件的数据(如 API 响应或文件内容)
  • Tools:LLM 可以调用的函数(经用户批准)
  • Prompts:帮助用户完成特定任务的预先编写的模板

这里我们使用 Python + Cline 实现MCP Server的编码与调用,Python版本为3.10或更高,MCP SDK 版本为 1.2.0 或更高。

官方文档建议使用 uv 作为包管理工作,但是我偏不用,我就用相对原始的 pip 命令。

这里默认各位朋友是掌握了 Python 语言的,所以我省略了环境的配置,也省略了项目的创建。我的项目长这样:

在这里插入图片描述

对,基本没啥东西,就一个 main.py 文件。

我们先 mock 一些数据,用于后续使用。实际开发场景中,我们会去访问真实数据。

# 准备模拟数据
# 用户信息数组,用户信息key是用户名,value是用户信息
users = {"alice": {"name": "Alice", "age": 25},"bob": {"name": "Bob", "age": 30},"charlie": {"name": "Charlie", "age": 35},
}# 用户资源数组,用户资源key是用户名,value是资源内容
resources = {"alice": ["res1", "res2", "res3"],"bob": ["res4", "res5"],"charlie": ["res6"]
}# 应用配置信息
config = {"app_name": "FastMCP","app_version": "1.0.0","app_description": "A FastMCP server","app_author": "FastMCP Team","app_license": "MIT","app_url": "https://github.com/FastMCP/FastMCP","app_contact": "https://github.com/FastMCP/FastMCP/issues","app_email": "fastmcp@gmail.com",
}

构建 MCP 服务,官方 MCP 包中提供了 FastMCP 类用于快速创建 MCP 服务。

mcp: FastMCP = FastMCP(name="这是FastMCP的参数name",instructions="""这是FastMCP的参数instructions."""
)

根据官方文档介绍,我们使用 @mcp.tool() 装饰器定义 Tools,使用@mcp.resource() 定义 Resources,使用 @mcp.prompt() 定义 Prompts。

#             @server.tool()
#             def my_tool(x: int) -> str:
#                 return str(x)
#
#             @server.tool()
#             def tool_with_context(x: int, ctx: Context) -> str:
#                 ctx.info(f"Processing {x}")
#                 return str(x)
#
#             @server.tool()
#             async def async_tool(x: int, context: Context) -> str:
#                 await context.report_progress(50, 100)
#                 return str(x)
@mcp.tool()
def custom_tool(name: str) -> str:"""获取指定用户的信息"""return json.dumps(users[name])#             @server.resource("resource://my-resource")
#             def get_data() -> str:
#                 return "Hello, world!"
#
#             @server.resource("resource://my-resource")
#             async get_data() -> str:
#                 data = await fetch_data()
#                 return f"Hello, world! {data}"
#
#             @server.resource("resource://{city}/weather")
#             def get_weather(city: str) -> str:
#                 return f"Weather for {city}"
#
#             @server.resource("resource://{city}/weather")
#             async def get_weather(city: str) -> str:
#                 data = await fetch_weather(city)
#                 return f"Weather for {city}: {data}"
@mcp.resource("resource://{user_name}")
def custom_resource_template(user_name: str) -> str:"""获取指定用户的资源列表"""return json.dumps(resources[user_name])@mcp.resource("config://app")
def custom_resource() -> str:"""获取应用配置信息"""# 将config转换成json字符串return json.dumps(config)#             @server.prompt()
#             def analyze_table(table_name: str) -> list[Message]:
#                 schema = read_table_schema(table_name)
#                 return [
#                     {
#                         "role": "user",
#                         "content": f"Analyze this schema:\n{schema}"
#                     }
#                 ]
#
#             @server.prompt()
#             async def analyze_file(path: str) -> list[Message]:
#                 content = await read_file(path)
#                 return [
#                     {
#                         "role": "user",
#                         "content": {
#                             "type": "resource",
#                             "resource": {
#                                 "uri": f"file://{path}",
#                                 "text": content
#                             }
#                         }
#                     }
#                 ]
@mcp.prompt()
def custom_prompt(name: str) -> str:"""欢迎指定用户"""return f"欢迎,{name}!有什么需要帮助的码?"

定义运行参数,运行我们的 MCP 服务

if __name__ == "__main__":# Initialize and run the servermcp.run()

完整代码如下:

import jsonfrom mcp.server import FastMCP# 准备模拟数据
# 用户信息数组,用户信息key是用户名,value是用户信息
users = {"alice": {"name": "Alice", "age": 25},"bob": {"name": "Bob", "age": 30},"charlie": {"name": "Charlie", "age": 35},
}# 用户资源数组,用户资源key是用户名,value是资源内容
resources = {"alice": ["res1", "res2", "res3"],"bob": ["res4", "res5"],"charlie": ["res6"]
}# 应用配置信息
config = {"app_name": "FastMCP","app_version": "1.0.0","app_description": "A FastMCP server","app_author": "FastMCP Team","app_license": "MIT","app_url": "https://github.com/FastMCP/FastMCP","app_contact": "https://github.com/FastMCP/FastMCP/issues","app_email": "fastmcp@gmail.com",
}mcp: FastMCP = FastMCP(name="这是FastMCP的参数name",instructions="""这是FastMCP的参数instructions."""
)#             @server.tool()
#             def my_tool(x: int) -> str:
#                 return str(x)
#
#             @server.tool()
#             def tool_with_context(x: int, ctx: Context) -> str:
#                 ctx.info(f"Processing {x}")
#                 return str(x)
#
#             @server.tool()
#             async def async_tool(x: int, context: Context) -> str:
#                 await context.report_progress(50, 100)
#                 return str(x)
@mcp.tool()
def custom_tool(name: str) -> str:"""获取指定用户的信息"""return json.dumps(users[name])#             @server.resource("resource://my-resource")
#             def get_data() -> str:
#                 return "Hello, world!"
#
#             @server.resource("resource://my-resource")
#             async get_data() -> str:
#                 data = await fetch_data()
#                 return f"Hello, world! {data}"
#
#             @server.resource("resource://{city}/weather")
#             def get_weather(city: str) -> str:
#                 return f"Weather for {city}"
#
#             @server.resource("resource://{city}/weather")
#             async def get_weather(city: str) -> str:
#                 data = await fetch_weather(city)
#                 return f"Weather for {city}: {data}"
@mcp.resource("resource://{user_name}")
def custom_resource_template(user_name: str) -> str:"""获取指定用户的资源列表"""return json.dumps(resources[user_name])@mcp.resource("config://app")
def custom_resource() -> str:"""获取应用配置信息"""# 将config转换成json字符串return json.dumps(config)#             @server.prompt()
#             def analyze_table(table_name: str) -> list[Message]:
#                 schema = read_table_schema(table_name)
#                 return [
#                     {
#                         "role": "user",
#                         "content": f"Analyze this schema:\n{schema}"
#                     }
#                 ]
#
#             @server.prompt()
#             async def analyze_file(path: str) -> list[Message]:
#                 content = await read_file(path)
#                 return [
#                     {
#                         "role": "user",
#                         "content": {
#                             "type": "resource",
#                             "resource": {
#                                 "uri": f"file://{path}",
#                                 "text": content
#                             }
#                         }
#                     }
#                 ]
@mcp.prompt()
def custom_prompt(name: str) -> str:"""欢迎指定用户"""return f"欢迎,{name}!有什么需要帮助的码?"if __name__ == "__main__":# Initialize and run the servermcp.run()

我们在命令行输入:mcp dev main.py,回车执行

在这里插入图片描述

打开这个地址,这个网页就是 MCP Inspector,我们可以用来调试我们开发的 MCP服务。

在这里插入图片描述

这里我们需要将 Command 中的uv替换成 D:\Demos\mcp-server-demo\mcp-server-demo\venv\Scripts\python.exe,将 Arguments 换成 main.py,接着点击 Connect 连接我们的 MCP 服务。为什么我们要换一下这里的 Command 和 Arguments???因为我们压根就没用到 uv,这里的本质还是运行我们自己编写的main.py 文件,希望各位朋友能理解。怎么运行不重要,能运行起来才重要~

MCP Inspector 不是这篇文章的讨论和使用重点,这里就放一张图各位看看效果,MCP Inspector 主要还是用来调试我们自己的 MCP Server。

在这里插入图片描述

在 MCP Client中使用,这里我们选择的是 Cline。Cline 是什么、怎么使用这里不再赘述,感兴趣可以看我之前的文章,里面有详细的介绍和使用。

DeepSeek+Cline使用教程(告别手动编码,全面拥抱AI)

这里我们配置好MCP服务,配置如下:

{"mcpServers": {"mcp-server-demo": {"command": "D:Demos\\mcp-server-demo\\mcp-server-demo\\venv\\Scripts\\python","args": ["D:\\Demos\\mcp-server-demo\\mcp-server-demo\\main.py"],"autoApprove": ["custom_tool"]}}
}

完整配置效果如下图,有绿色小圆点表示配置成功。

在这里插入图片描述

配置成功后,我们可以简单试下效果。

“获取alice的用户信息”

在这里插入图片描述

“获取应用的配置信息”

在这里插入图片描述

05 总结

至此,我们就从 0~1 完成了一个简单的,属于我们自己的 MCP Server。

整个问答的幕后逻辑如下,当我们提出问题时:

  1. 客户将您的问题发送给 LLM
  2. LLM 分析了可用的工具并决定使用哪一个
  3. 客户端(Cline)通过 MCP 服务器执行所选工具
  4. 将结果发送回给 LLM
  5. LLM 制定自然语言响应
  6. 响应向使用者显示

这生活已经很难了,难能可贵的是您能在百忙之中抽空阅读这些文章。如果能给到您一点小小的帮助,也是我非常喜闻乐见的😁

相关文章:

  • class com.alibaba.fastjson.JSONObject cannot be cast to class
  • 安装WSL2.0
  • 计算机组成与体系结构:内存层次结构(Memory Hierarchy)
  • Kotlin协程学习笔记
  • 无人机飞控运行在stm32上的RTOS实时操作系统上,而不是linux这种非实时操作系统的必要性
  • 微服务架构下数据库范式的失效与反范式设计的崛起
  • wps表格保存时提示上传错误报0XFFF40005错误,提示撤销之前操作或者另存为
  • JavaScript 渲染内容爬取:Puppeteer 入门
  • 线性DP:最短编辑距离
  • 2025.04.20【Lollipop】| Lollipop图绘制命令简介
  • 实验四 Java图形界面与事件处理
  • C++学习:六个月从基础到就业——内存管理:智能指针详解
  • 集成学习实际案例
  • 储能集装箱电池簇安装支架结构设计(大纲)
  • 代码随想录第22天:回溯算法4
  • 从规则到大模型:知识图谱信息抽取实体NER与关系RE任务近10年演进发展详解
  • 项目班——0419——functionbind生产消费(未完成)
  • 如何测试雷达与相机是否时间同步?
  • list的学习
  • Missashe考研日记-day23
  • 花3000元就能买“国际机构”的证书?揭秘假证产业链
  • 在没有穹顶的剧院,和春天的音乐会来一场约会
  • 能源央企资产重组大提速,专业化整合掀起新热潮
  • 新消费观察 | 重点深耕,外资科技企业继续看好中国发展
  • 杭州挂牌3宗住宅用地,起始总价约30.04亿元
  • 蓝思科技一季度净利增近四成,预计关税对整体经营影响非常有限