Schwertlilien
As a recoder: notes and ideas.

2025-8-23

如何向GPT老师提问?

modelcontextprotocol.io/llms-full.txt

描述你的服务器

提供文档后,请向 Claude 清楚地描述你想要构建的服务器类型。 详细说明:

  1. 你的服务器将公开哪些资源(resources)
  2. 它将提供哪些工具(tools)
  3. 它应该提供哪些提示(prompts)
  4. 它需要与哪些外部系统交互

在开发 MCP 服务器时,

我需要注意的:

  1. 首先从核心功能开始,然后迭代添加更多功能
  2. 要求 Claude 解释你无法理解的代码部分
  3. 根据需要请求修改或改进
  4. 让 Claude 帮助你测试服务器并处理极端情况

LLM需要注意的:

  1. 将复杂的服务器分解为更小的部分
  2. 在继续之前彻底测试每个组件
  3. 牢记安全性——验证输入并适当限制访问
  4. 充分记录你的代码,以便将来维护
  5. 仔细遵循 MCP 协议规范

MCP概念

核心架构

协议层(Protocol Layer)

功能:处理消息组帧、请求/响应连接、以及高级通信方式。

  1. send_request: 发送请求数据
  2. send_notification: 发现单项通知
  3. _received_request: 处理C/S发来的请求
  4. _received_notification: 被动处理来自对方的通知
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class Session(BaseSession[RequestT, NotificationT, ResultT]):
async def send_request(
self,
request: RequestT,
result_type: type[Result]
) -> Result:
"""
Send request and wait for response. Raises McpError if response contains error.
"""
# Request handling implementation

async def send_notification(
self,
notification: NotificationT
) -> None:
"""Send one-way notification that doesn't expect response."""
# Notification handling implementation

async def _received_request(
self,
responder: RequestResponder[ReceiveRequestT, ResultT]
) -> None:
"""Handle incoming request from other side."""
# Request handling implementation

async def _received_notification(
self,
notification: ReceiveNotificationT
) -> None:
"""Handle incoming notification from other side."""
# Notification handling implementation

传输层(Transport Layer)

  1. 标准输入/输出传输(Stdio Transport)

    • 使用标准输入/输出进行通信
    • 适用于本地进程
  2. HTTP with SSE 传输

    • 使用服务器发送事件(Server-Sent Events,SSE)进行服务器到客户端的消息传递
    • 使用 HTTP POST 进行客户端到服务器的消息传递

消息类型

主要消息类型 说明 代码
请求(Requests) 期望另一方做出响应 interface Request { method: string; params?: { ... }; }
结果(Results) 对请求的成功响应 interface Result { [key: string]: unknown; }
错误(Errors) 请求失败 interface Error { code: number; message: string; data?: unknown; }
通知(Notifications) 不需要响应的单向消息 interface Notification { method: string; params?: { ... }; }

C/S连接

  1. 客户端发送带有协议版本和功能的 initialize 请求
  2. 服务器响应其协议版本和功能
  3. 客户端发送 initialized 通知作为确认
  4. 正常的 message exchange (消息交换) 开始

消息交换

  • 请求-响应(Request-Response):客户端或服务器发送请求,另一方响应
  • 通知(Notifications):任一方发送单向消息

终止

任一方都可以终止连接:

  • 通过 close() 清理关闭
  • 传输断开连接
  • 错误情况

错误处理(Error Handling)

MCP 定义了以下标准错误代码:

1
2
3
4
5
6
7
8
enum ErrorCode {
// Standard JSON-RPC error codes
ParseError = -32700,
InvalidRequest = -32600,
MethodNotFound = -32601,
InvalidParams = -32602,
InternalError = -32603
}

SDK 和应用程序可以在 -32000 以上定义自己的错误代码。错误通过以下方式传播:

  • 对请求的错误响应
  • 传输上的错误事件
  • 协议级别的错误处理程序

示例

下面的代码实现了一个最小化的 MCP 服务器,功能是:

  1. 通过 STDIO 传输层(标准输入输出)与客户端通信。
  2. 向客户端暴露一个名为 Example Resource 的资源(URI 为 example://resource)。
  3. 持续监听客户端的请求(如 “列出所有资源”),并返回对应的资源列表。

客户端连接该服务器后,可以调用 list_resources 方法获取这个资源的信息,是 MCP 协议 “资源共享” 能力的基础示例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import asyncio  # 用于处理异步操作(MCP 服务器基于异步通信)
import mcp.types as types # 导入 MCP 协议定义的类型(如资源、工具等数据结构)
from mcp.server import Server # 导入 MCP 服务器基类(核心骨架)
from mcp.server.stdio import stdio_server # 导入基于标准输入输出的服务器传输层

app = Server("example-server") # 创建MCP Server实例

@app.list_resources() # 注册资源列表接口
async def list_resources() -> list[types.Resource]:
return [
types.Resource(
uri="example://resource", # 资源的唯一标识(类似 URL),客户端可通过此 URI 访问该资源。
name="Example Resource" # 资源的名称(用于展示给用户或客户端)。
)
]

async def main():
async with stdio_server() as streams: # 启动基于 STDIO 的传输层
# STDIO 传输适用于本地进程间通信(如同一台机器上的客户端和服务器)。
# streams:包含两个元素的元组 (reader, writer),
# -- 分别是异步读取流(接收客户端请求)和写入流(发送响应给客户端)。
await app.run(
streams[0], # STDIO 传输层的读取流(用于接收客户端的请求数据)。
streams[1], # STDIO 传输层的写入流(用于向客户端发送响应数据)。
app.create_initialization_options() # 创建服务器初始化参数(MCP 协议要求的启动配置,如服务器名称、支持的协议版本等)。
)

if __name__ == "__main__":
asyncio.run(main)

针对示例-我的疑问

1.@mcp.list_resources()这个和@mcp.tools()有什么区别?二者是什么关系?

— 资源 v.s. 工具

资源(Resource) 工具(Tool)
资源是服务器对外暴露的 “可识别、可定位的实体”,比如 “天气服务资源”“用户数据资源”“文件存储资源” 等。每个资源通常用 uri(唯一标识,如 example://weather)和 name(友好名称,如 “天气查询服务”)来描述,是客户端感知服务器 “能力边界” 的基础元信息。 工具是服务器端实现的 “具体业务逻辑函数”,比如 “查询北京实时天气”“计算两个数的和”“读取本地文件” 等。客户端可以主动调用这些工具来完成实际任务,工具是服务器 “业务能力” 的直接体现。

— list_resources() v.s. tools()

对比维度 @mcp.list_resources() @mcp.tools()
核心用途 用于向客户端暴露服务器的「资源列表」,回答“服务器有什么” 用于向客户端暴露服务器的「可调用工具 / 功能」,回答“服务器能做什么”
装饰的函数职责 函数需返回 list[types.Resource](资源列表),仅负责“描述资源”,不包含业务逻辑 函数是实际业务逻辑的实现(如查询、计算、调用外部 API),客户端调用时会执行该函数
返回值/输出 输出「资源元数据」(如 uriname),用于客户端“能力发现” 输出「工具执行结果」(如天气数据、计算结果),用于客户端完成具体任务
面向的交互阶段 客户端与服务器初始化连接时(如客户端首次连接,先获取资源列表) 客户端与服务器正常交互时(如客户端需要执行任务,主动调用工具)

@app.list_resources()@app.tools() 不是“二选一”的关系,而是 MCP 服务器向客户端暴露能力的“两层协作体系”,具体关系可概括为 2 点:

一、资源(Resource)可作为工具(Tool)的“组织容器”

MCP 中,资源通常是工具的“归类载体”——一个资源可以包含多个相关工具,形成“资源-工具”的层级关系,让客户端更容易理解和调用。
例如:用 @app.list_resources() 注册一个“天气服务资源”:

1
2
3
4
5
6
7
8
@app.list_resources()
async def list_resources() -> list[types.Resource]:
return [
types.Resource(
uri="example://weather-service", # 资源唯一标识
name="天气查询服务" # 资源友好名称
)
]

再用 @app.tools() 为这个资源注册 2 个工具(绑定到该资源的 uri):

1
2
3
4
5
6
7
8
9
@app.tools(resource_uri="example://weather-service")  # 绑定到“天气服务资源”
async def get_realtime_weather(city: str) -> dict:
# 实际业务逻辑:调用天气API查询实时天气
return {"city": city, "temperature": 25, "status": "sunny"}

@app.tools(resource_uri="example://weather-service")
async def get_forecast_weather(city: str, days: int) -> list:
# 实际业务逻辑:查询未来N天天气预报
return [{"date": "2024-08-25", "temperature": 26, "status": "cloudy"}, ...]

此时,客户端先通过 list_resources() 发现“天气查询服务”这个资源,再进一步获取该资源下的 get_realtime_weatherget_forecast_weather 两个工具,避免工具混乱。

二、交互流程上:“资源发现”先于“工具调用”

在 MCP 客户端与服务器的正常交互中,二者通常遵循“先发现、后调用”的逻辑:

  1. 客户端连接服务器后,首先调用 list_resources() 对应的接口:获取服务器的所有资源元信息,知道“服务器有哪些可用服务(资源)”;
  2. 客户端根据资源选择需要的工具:比如想查天气,就找到“天气查询服务”资源对应的工具列表;
  3. 客户端调用具体的 tools:执行实际任务(如调用 get_realtime_weather("北京"))。

如果没有 list_resources(),客户端无法感知服务器的资源边界,可能不知道该调用哪些工具;如果没有 toolslist_resources() 注册的资源只是“空壳元信息”,无法完成任何实际任务。

2.C&S?

通过 STDIO 传输层(标准输入输出)与客户端通信。 向客户端暴露一个名为 Example Resource 的资源(URI 为 example://resource)。 持续监听客户端的请求(如 “列出所有资源”),并返回对应的资源列表。 客户端连接该服务器后,可以调用 list_resources 方法获取这个资源的信息,是 MCP 协议 “资源共享” 能力的基础示例。

这个最小化 MCP 服务器,就像一家“刚开业的小店”:

  1. 它只做“告诉顾客有啥服务”(暴露资源),不做复杂业务;
  2. 它只能“面对面沟通”(STDIO 传输),不能远程服务;
  3. 它是后续扩展的基础——只要在这个框架上加 @app.tools(),就能让“服务清单”变成“能实际办事的服务”。

MCP 协议的“资源共享”,本质就是“让服务端清晰地告诉客户端‘我有什么’,让客户端不用瞎猜就能找到可用的能力”,而 STDIO 就是二者“近距离沟通的方式”。

把 MCP 服务器想象成「社区里的一家便民小店」,客户端想象成「来店里办事的顾客」,核心逻辑对应如下:

技术概念 生活场景类比 通俗解释
MCP 服务器 社区便民小店 店里有固定服务/商品,等着顾客上门请求
STDIO 传输层 小店的“面对面窗口” 顾客和店员只能在窗口前说话(不能打电话/线上沟通),同一间屋子才能交互
资源(Resource) 小店的“服务清单”(如“代收快递”“复印”) 告诉顾客“店里能提供什么”,是服务的“名字和编号”
list_resources() 方法 顾客问“老板,你家能办啥业务?” 店员递出一张写满服务的清单,回答顾客的疑问
客户端调用 list_resources() 顾客主动问业务清单 顾客上门后,第一步先确认店里有没有自己需要的服务

用“小店营业”的流程,对应解释

假设我们要开这家“MCP 小店”(运行代码),顾客(客户端)上门办事,完整流程如下,你可以跟着动手操作(需要提前安装 mcp 库:pip install mcp):

1. 开店(运行 MCP 服务器代码)

先把你之前的代码保存为 mcp_server.py,就像“小店开门营业”:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# mcp_server.py(就是你之前的代码)
import asyncio
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server

# 1. 开一家叫“example-server”的小店
app = Server("example-server")

# 2. 定义“小店的服务清单”(只有1项服务:Example Resource)
@app.list_resources()
async def list_resources() -> list[types.Resource]:
return [
types.Resource(
uri="example://resource", # 服务的唯一编号(比如“小店代收快递的编号001”)
name="Example Resource" # 服务的友好名字(比如“代收快递服务”)
)
]

# 3. 打开“面对面窗口”(STDIO),等着顾客上门
async def main():
async with stdio_server() as streams:
await app.run(
streams[0],
streams[1],
app.create_initialization_options()
)

if __name__ == "__main__":
asyncio.run(main()) # 这里注意!原代码少了括号,正确是 asyncio.run(main())

运行服务器:打开一个命令行窗口(相当于“小店开门”),输入命令:

1
python mcp_server.py

此时窗口会“卡住”——这不是报错,而是小店“窗口打开,等着顾客说话”(服务器监听客户端请求)。

2. 顾客上门(写 MCP 客户端)

再写一个简单的客户端代码 mcp_client.py,相当于“顾客上门,问老板有啥服务”:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# mcp_client.py(顾客的“办事脚本”)
import asyncio
from mcp.client import ClientSession
from mcp.client.stdio import stdio_client
from mcp.parameters import StdioServerParameters

async def main():
# 1. 顾客走到小店窗口前(连接服务器的STDIO传输层)
# 这里的 command 是“启动服务器的命令”(因为STDIO是本地交互,需要知道服务器在哪)
server_params = StdioServerParameters(
command="python", # 启动服务器的工具(python)
args=["mcp_server.py"] # 服务器脚本路径
)

# 2. 打开和小店的“对话通道”(STDIO连接)
async with stdio_client(server_params) as (reader, writer):
# 3. 和小店建立正式会话(相当于“顾客说‘您好,我要办事’”)
async with ClientSession(reader, writer) as session:
# 4. 顾客问:“老板,你家能办啥业务?”(调用list_resources())
resources_result = await session.list_resources()

# 5. 顾客拿到服务清单,读出来
print("从小店(服务器)拿到的服务清单:")
for resource in resources_result.resources:
print(f"- 服务名字:{resource.name}")
print(f"- 服务编号(URI):{resource.uri}")
print("-" * 20)

if __name__ == "__main__":
asyncio.run(main())

3. 顾客办事(运行客户端,看实际结果)

再打开一个新的命令行窗口(相当于“顾客走到小店门口”),输入命令运行客户端:

1
python mcp_client.py

你会看到客户端窗口输出:

1
2
3
4
从小店(服务器)拿到的服务清单:
- 服务名字:Example Resource
- 服务编号(URI):example://resource
--------------------

同时,之前“卡住”的服务器窗口会默默记录这次交互(如果加日志的话能看到)——这就是一次完整的“客户端请求→服务器响应”。

“拿到服务清单后,顾客还能干嘛?”你可能会问:“顾客只拿清单有啥用?不能真办事啊!”
这就是“最小化服务器”的意义——它只做“暴露资源”这一件基础事,就像小店刚开业,先告诉顾客“我能办啥”,后续可以再加“实际办事的工具”(对应 @app.tools() 装饰器)。

比如,我们给小店加一个“用 Example Resource 服务复印文件”的工具(修改服务器代码 mcp_server.py):

1
2
3
4
5
6
7
8
9
# 在原服务器代码基础上,新增工具定义
@app.tools(resource_uri="example://resource") # 把工具绑定到“Example Resource”服务
async def copy_file(file_path: str, copy_count: int) -> dict:
"""实际办事的工具:根据服务编号(URI),提供“复印文件”服务"""
return {
"status": "成功",
"message": f"用「Example Resource」服务复印了 {copy_count} 份文件",
"file_path": file_path
}

此时客户端拿到“服务清单”后,还能进一步调用 copy_file 工具(比如“复印 test.txt 文件3份”),服务器会返回实际的复印结果。

搜索
匹配结果数:
未搜索到匹配的文章。