【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。
整个问答的幕后逻辑如下,当我们提出问题时:
- 客户将您的问题发送给 LLM
- LLM 分析了可用的工具并决定使用哪一个
- 客户端(Cline)通过 MCP 服务器执行所选工具
- 将结果发送回给 LLM
- LLM 制定自然语言响应
- 响应向使用者显示
这生活已经很难了,难能可贵的是您能在百忙之中抽空阅读这些文章。如果能给到您一点小小的帮助,也是我非常喜闻乐见的😁