我要评论模型上下文协议（mcp, model context protocol）是 anthropic 在2024年提出的一种开放标准协议，用于标准化 ai 模型与外部工具和数据源的集成方式。
可以将 mcp 类比为 ai 世界中的 “usb-c 接口”：它为大型语言模型（llm）连接各种工具和数据源提供了一种统一的方法。
mcp 使用 json-rpc 2.0 作为消息格式，在客户端和服务器之间传递请求和响应。
本示例将展示如何使用 python 实现一个最基础的 mcp 协议，包括 mcp 服务器和 mcp 客户端两部分。
我们将支持
- discovery（发现）
- invoke（调用）
- retrieve（获取）
等基本操作，并通过一个简单的“计算器”工具（支持加法和乘法）演示协议的工作原理。
背景和设计概述
mcp 协议采用 客户端-服务器 架构。
mcp服务器提供一组工具（tools）或资源（resources），mcp客户端可以发现服务器提供的功能并进行调用 。
两者之间通过json-rpc进行通信，以标准的请求/响应消息交换指令和数据。
按照 mcp 规范：
发现（discovery）：客户端能够查询服务器，获取其提供的工具列表、资源列表等。这通常通过调用 tools/list 或 resources/list 等方法实现。
调用（invoke）：客户端可以请求执行服务器上的某个工具功能，例如调用计算器的加法或乘法操作。规范中约定使用 tools/call 方法来调用指定名称的工具，并传递所需参数。
获取（retrieve）：客户端能够检索数据内容，例如获取某个资源的具体内容。规范中提供了如 resources/read 等方法用于检索资源内容。
在简单工具调用场景下，调用的结果会直接作为响应返回；但对于长任务或资源内容，常采用单独的检索步骤获取结果。
实现思路
我们将使用 python 标准库构建一个简易的 mcp 服务器和客户端。
服务器会注册至少一个工具——一个简单计算器，包含加法和乘法 功能。
客户端可以通过发现操作获取服务器上可用的工具列表，并通过调用操作执行这些工具函数并得到结果。
为了尽量减少依赖并简化部署，我们将直接使用 python 自带的 httpserver 来处理 json-rpc 请求，使用内置的 json 和 http.client 模块实现通信。
代码结构将力求清晰，方便日后扩展更多功能（例如增加新的工具、资源或更复杂的异步处理）。
主要功能模块：
mcpserver
处理客户端请求的服务器类。
维护可用的工具和资源列表，能够响应 tools/list、tools/call、resources/list、resources/read 等 json-rpc 方法。
mcpclient
客户端类或脚本，用于连接mcp服务器发送json-rpc请求。支持发现工具列表、调用指定工具、（可选）获取资源内容等操作。
通信格式
遵循json-rpc 2.0规范，每个请求包含 jsonrpc 版本、请求 id、方法名和参数，响应则包含对应的 id以及结果或错误信息。例如，列出工具的请求方法为 tools/list；
调用工具的请求方法为 tools/call，需要提供工具名称和参数。
本示例中，我们的实现将返回简化的结果格式，以便直观展示功能（实际mcp可能返回更结构化的内容，如带有类型说明的内容列表）。
下面我们分别给出 mcp 服务器和客户端的代码实现，并包含必要的注释和使用说明。
mcp服务器实现 (server)
首先，实现mcp服务器。我们创建一个 mcpserver 类用于注册工具和资源，并处理传入的请求。
然后使用 python 的 basehttprequesthandler 来构建http接口，使服务器能够通过http接收 json-rpc 请求。
计算器工具将作为示例工具注册到服务器上。
import json
from http.server import basehttprequesthandler, httpserver
# 定义一个tool类，用于存储工具的元数据和执行函数
class tool:
def __init__(self, name, description, input_schema, output_schema, func):
self.name = name
self.description = description
self.input_schema = input_schema # json schema for input parameters
self.output_schema = output_schema # json schema for output
self.func = func # function to execute the tool
# mcp服务器类，维护工具和资源，并处理json-rpc请求
class mcpserver:
def __init__(self):
# 注册工具列表：这里添加一个“计算器”工具，提供加法和乘法 功能
def add(a, b): # 简单加法函数
return a + b
def multiply(a, b): # 简单乘法函数
return a * b
# 为演示，将加法和乘法作为两个独立工具注册
self.tools = {
"add": tool(
name="add",
description="add two numbers",
input_schema={ # 输入参数的json schema定义
"type": "object",
"properties": {
"a": {"type": "number", "description": "first operand"},
"b": {"type": "number", "description": "second operand"}
},
"required": ["a", "b"]
},
output_schema={ # 输出结果的json schema定义
"type": "number"
},
func=add
),
"multiply": tool(
name="multiply",
description="multiply two numbers",
input_schema={
"type": "object",
"properties": {
"a": {"type": "number", "description": "first operand"},
"b": {"type": "number", "description": "second operand"}
},
"required": ["a", "b"]
},
output_schema={
"type": "number"
},
func=multiply
)
}
# 可选：注册资源列表，这里添加一个简单文本资源示例
self.resources = {
"mem://greeting": {
"name": "greeting.txt",
"title": "greeting message",
"description": "a welcome text message",
"mimetype": "text/plain",
"text": "hello, welcome to the mcp demo!" # 资源内容
}
}
def handle_request(self, request):
"""
处理json-rpc请求，根据method调用对应的处理逻辑。
返回json可序列化的响应字典。
"""
# 基础的json-rpc字段解析
jsonrpc = request.get("jsonrpc")
req_id = request.get("id")
method = request.get("method")
params = request.get("params", {})
# 确保符合 json-rpc 2.0 协议
if jsonrpc != "2.0" or req_id is none or method is none:
# 返回 json-rpc 错误：-32600 = invalid request
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32600, "message": "invalid request"}
}
# 处理方法：tools/list, tools/call, resources/list, resources/read
try:
if method == "tools/list":
# 列出可用工具
tools_info = []
for tool in self.tools.values():
tools_info.append({
"name": tool.name,
"description": tool.description,
"inputschema": tool.input_schema,
"outputschema": tool.output_schema
})
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"tools": tools_info
}
}
elif method == "tools/call":
# 调用指定名称的工具
name = params.get("name")
arguments = params.get("arguments", {})
if not name or name not in self.tools:
# 工具名称不存在
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32601, "message": f"tool '{name}' not found"}
}
# 获取工具并执行
tool = self.tools[name]
try:
# 调用工具函数，传入参数
result_value = tool.func(**arguments)
except typeerror as e:
# 参数不正确或缺失
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32602, "message": f"invalid parameters: {str(e)}"}
}
except exception as e:
# 工具执行过程中出现其他异常
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32000, "message": f"tool execution error: {str(e)}"}
}
# 返回执行结果（这里结果直接是数值，符合 outputschema 定义）
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"output": result_value # 将结果包装在一个字段中
}
}
elif method == "resources/list":
# 列出可用资源
resources_info = []
for uri, res in self.resources.items():
resources_info.append({
"uri": uri,
"name": res.get("name"),
"title": res.get("title"),
"description": res.get("description"),
"mimetype": res.get("mimetype")
})
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"resources": resources_info
}
}
elif method == "resources/read":
# 读取指定uri的资源内容
uri = params.get("uri")
if not uri or uri not in self.resources:
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32602, "message": f"invalid resource uri: {uri}"}
}
res = self.resources[uri]
# 返回资源内容（文本）
return {
"jsonrpc": "2.0",
"id": req_id,
"result": {
"contents": [
{
"uri": uri,
"name": res.get("name"),
"title": res.get("title"),
"mimetype": res.get("mimetype"),
"text": res.get("text")
}
]
}
}
else:
# 未知的方法
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32601, "message": f"method '{method}' not found"}
}
except exception as e:
# 捕获任何未预料的异常，返回内部错误
return {
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32603, "message": f"internal error: {str(e)}"}
}
# 实例化 mcpserver，全局使用
mcp_server = mcpserver()
# 定义http请求处理器类
class mcprequesthandler(basehttprequesthandler):
def do_post(self):
# 读取请求体中的json数据
content_length = int(self.headers.get('content-length', 0))
request_bytes = self.rfile.read(content_length)
try:
request_json = json.loads(request_bytes.decode('utf-8'))
except json.jsondecodeerror:
# 如果不是合法的json，返回错误
response = {
"jsonrpc": "2.0",
"id": none,
"error": {"code": -32700, "message": "parse error"}
}
self._send_response(response)
return
# 调用mcpserver处理请求，获取响应
response = mcp_server.handle_request(request_json)
# 将响应发送回客户端
self._send_response(response)
def _send_response(self, response_obj):
"""辅助方法：发送json响应。"""
response_bytes = json.dumps(response_obj).encode('utf-8')
# 构造http响应头
self.send_response(200)
self.send_header("content-type", "application/json; charset=utf-8")
self.send_header("content-length", str(len(response_bytes)))
self.end_headers()
# 写出响应内容
self.wfile.write(response_bytes)
self.wfile.flush()
# 启动 http 服务器（监听 localhost:8000）
if __name__ == "__main__":
server_address = ('', 8000) # 监听在所有接口的8000端口
httpd = httpserver(server_address, mcprequesthandler)
print("mcp server is running at http://localhost:8000")
print("available tools:", [name for name in mcp_server.tools.keys()])
httpd.serve_forever()上面的服务器代码说明：
工具注册：
在 mcpserver.init 中，我们定义了两个简单的算术函数 add 和 multiply，并使用它们分别注册了名为 "add" 和 "multiply" 的 tool。
每个 tool 包含名称、描述、输入参数模式 (input_schema)、输出结果模式 (output_schema) 以及执行函数 func。
在输入/输出模式中，我们使用了简化的 json schema 来描述参数（要求提供两个数字参数 a 和 b）和返回值类型（数字）。
这些元数据会在工具发现时提供给客户端，方便客户端或模型了解如何调用该工具。
资源注册：
为了演示 retrieve 操作，我们可选地注册了一个简单资源 mem://greeting，表示一段文本内容（模拟例如文件或数据库中的数据）。该资源包含一个 uri 标识符和元数据（名称、描述、mime类型等），以及文本内容 "hello, welcome to the mcp demo!"。资源将用于演示客户端如何列出并读取服务器上的数据内容。
请求处理：
mcpserver.handle_request 方法根据传入请求的 method 字段执行
相应逻辑：
- tools/list：返回服务器上所有可用工具的列表，包含每个工具的名称、描述、输入/输出模式等信息。
- tools/call：调用指定的工具。请求需提供参数 name（工具名）和 arguments（参数字典）。服务器检查工具是否存在，参数是否正确，然后调用对应的函数并获取结果。
结果在此示例中直接返回为一个简单数值，包装在 result.output 字段中。如出现错误（未知工具、参数错误、执行异常等），则返回 json-rpc 标准的 error 对象，包括错误码和信息。
resources/list：
返回可用资源列表，每项包含资源的 uri、名称、描述等元数据。
resources/read：
根据传入的 uri 参数，返回对应资源的内容。
在响应中，我们将内容放在 contents 列表中，每个元素包含资源的 meta 信息以及实际内容文本。
未知方法：
如果收到非上述定义的方法，返回 json-rpc 的 “method not found” 错误（错误码 -32601）。
http服务器：
使用 basehttprequesthandler 实现http接口。我们重载了 do_post 方法来处理 http post 请求（json-rpc 通常使用 post 来发送请求）。
服务器从请求体读取 json 数据并解析，然后调用 mcp_server.handle_request 获得结果，最后通过 _send_response 返回 json 响应。
httpserver 监听在 localhost:8000 端口，启动后打印可用工具列表以供参考。
注意，本实现为了简洁未包含认证、权限控制等机制，也未实现 mcp 规范中的会话管理、通知推送等高级功能。这是一个最基础的示例，展示了 json-rpc 通信和工具/资源调用的基本流程。在实际应用中，可能需要扩展支持例如能力协商（capability negotiation）、异步调用（如长时间运行任务的进度和结果获取）、安全认证等特性。
mcp客户端实现 (client)
接下来，实现与上述服务器交互的 mcp 客户端。客户端将按照顺序演示以下操作：
- 发现工具：请求获取服务器提供的工具列表（调用 tools/list）。
- 调用工具：调用加法和乘法工具（分别调用 tools/call，传入相应参数）。
- 发现资源：请求获取服务器提供的资源列表（调用 resources/list，可选步骤）。
- 获取资源内容：读取指定资源内容（调用 resources/read，示例读取 mem://greeting 文本）。
我们使用 python 内置的 http.client 模块发送 http post 请求，获取并解析 json 响应。以下是客户端代码：
import json
import http.client
# mcp客户端辅助函数：发送json-rpc请求并返回解析后的结果或错误
def send_json_rpc(method, params=none, request_id=1):
conn = http.client.httpconnection("localhost", 8000)
request_obj = {
"jsonrpc": "2.0",
"id": request_id,
"method": method,
"params": params or {}
}
# 将请求对象序列化为json并发送
conn.request("post", "/", body=json.dumps(request_obj), headers={
"content-type": "application/json"
})
# 获取http响应并解析json
response = conn.getresponse()
data = response.read().decode('utf-8')
conn.close()
try:
return json.loads(data)
except json.jsondecodeerror:
print("invalid json response:", data)
return none
# 1. 发现可用工具列表
resp = send_json_rpc("tools/list", {}, request_id=1)
print("available tools:")
if "result" in resp:
for tool in resp["result"]["tools"]:
name = tool["name"]
desc = tool.get("description")
print(f" - {name}: {desc}")
else:
# 错误情况
print("error listing tools:", resp.get("error"))
# 2. 调用加法工具 (例如 3 + 5)
add_params = {"name": "add", "arguments": {"a": 3, "b": 5}}
resp2 = send_json_rpc("tools/call", add_params, request_id=2)
if "result" in resp2:
result_value = resp2["result"].get("output")
print(f"result of add(3, 5): {result_value}")
else:
print("error invoking add tool:", resp2.get("error"))
# 3. 调用乘法工具 (例如 4 * 7)
mul_params = {"name": "multiply", "arguments": {"a": 4, "b": 7}}
resp3 = send_json_rpc("tools/call", mul_params, request_id=3)
if "result" in resp3:
result_value = resp3["result"].get("output")
print(f"result of multiply(4, 7): {result_value}")
else:
print("error invoking multiply tool:", resp3.get("error"))
# 4. （可选）列出可用资源
resp4 = send_json_rpc("resources/list", {}, request_id=4)
print("available resources:")
if "result" in resp4:
for res in resp4["result"]["resources"]:
uri = res["uri"]
desc = res.get("description")
print(f" - {uri}: {desc}")
else:
print("error listing resources:", resp4.get("error"))
# 5. （可选）读取指定资源内容 (例如 mem://greeting)
res_params = {"uri": "mem://greeting"}
resp5 = send_json_rpc("resources/read", res_params, request_id=5)
if "result" in resp5:
contents = resp5["result"].get("contents", [])
if contents:
text = contents[0].get("text")
print(f"content of resource mem://greeting: {text}")
else:
print("error reading resource:", resp5.get("error"))
说明：
- send_json_rpc 函数封装了发送请求和接收响应的过程。它建立http连接到 localhost:8000，发送json格式的rpc请求，并返回解析后的python字典对象。
- 客户端按照步骤构造请求：
- tools/list 请求没有额外参数，返回服务器可用工具列表。在示例中，服务器应返回我们注册的 "add" 和 "multiply" 工具，各自的描述和参数模式也包含在结果中。
- tools/call 请求加法工具，提供参数 {a:3, b:5}。服务器返回结果，客户端从响应中提取 output 字段即计算结果（期望为8）。类似地调用乘法工具 4*7，应得到28。
- resources/list 请求返回服务器可用资源列表。在我们的示例中，应当包含 mem://greeting 这个资源及其描述信息。
- resources/read 请求读取 mem://greeting，服务器将返回其内容文本。客户端打印出该文本内容。
在打印输出中，我们对结果做了简单的格式化。例如：
available tools: - add: add two numbers - multiply: multiply two numbers result of add(3, 5): 8 result of multiply(4, 7): 28 available resources: - mem://greeting: a welcome text message content of resource mem://greeting: hello, welcome to the mcp demo!
这样，我们就完成了一个基本的 mcp 协议交互流程：客户端发现服务器的功能，并成功调用工具得到计算结果，还演示了资源的发现和获取。整个通信过程采用 json-rpc 格式，确保了请求和响应的标准化
使用说明
要运行该示例，请按照以下步骤操作：
- 启动服务器：将上述服务器代码保存为
mcp_server.py并运行。例如:
python mcp_server.py
服务器将启动并监听端口 8000。在控制台上可以看到启动日志，例如可用的工具列表打印:
mcp server is running at http://localhost:8000 available tools: ['add', 'multiply']
- 运行客户端：在服务器运行的同时，打开另一个终端窗口，将上述客户端代码保存为 mcp_client.py 并运行:
python mcp_client.py
客户端将依次发送 json-rpc 请求到服务器，并在终端打印收到的响应结果。
- 观察结果：你应当在客户端终端看到类似输出：
available tools: - add: add two numbers - multiply: multiply two numbers result of add(3, 5): 8 result of multiply(4, 7): 28 available resources: - mem://greeting: a welcome text message content of resource mem://greeting: hello, welcome to the mcp demo!
这些输出对应了我们在客户端代码中的打印语句，验证了各项操作成功执行。
- 交互验证：你也可以使用其他json-rpc客户端（例如curl或postman）手动发送请求进行测试。例如，使用 curl 获取工具列表：
curl -x post http://localhost:8000/ -h "content-type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'你将收到json响应，包含已注册工具的详细信息。
扩展与总结
本示例提供了一个最基础的 mcp 协议实现框架。通过清晰的结构设计，扩展该实现非常方便：
增加新工具：
可以在服务器的 self.tools 字典中添加新的 tool 实例，并实现对应的函数。例如，可以添加一个字符串处理工具或数据库查询工具。客户端发现机制使得新工具能够被动态发现，无需更改客户端代码。
增加新资源：
同样地，可扩充 self.resources 字典来暴露新的资源数据。结合客户端的资源检索流程，可以为模型提供更多上下文数据。
复杂功能：
可考虑实现异步调用。如果某工具执行时间较长，可修改 tools/call 的实现为立即返回一个任务id，然后通过新增的 tools/retrieve（或类似方法）根据id获取结果。这类似于拆分调用和结果提取两个步骤，在长任务场景下很有用。
安全和认证：
实际应用中，应在传输层（如https）以及协议层增加身份验证和授权机制，确保只有获得许可的客户端才能访问敏感工具或数据。mcp 规范在最新版本中也引入了 oauth2 等认证支持。
日志和错误处理：
可以扩展错误日志记录、请求计量等，以提高可监控性和可靠性。
通过上述示例，我们验证了json-rpc风格的 mcp 通信模型：客户端可以动态发现工具并调用之，从而让llm应用具备即插即用的扩展能力。
虽然我们的实现是简化的，但它奠定了构建完整 mcp 服务器-代理系统的基础结构。
在未来，你可以逐步按照 mcp 规范增加prompts（预设提示模板）、更多工具类型以及高级特性，将其演变为功能完整的 mcp 服务。希望这个示例能帮助你理解 mcp 协议的核心机制，并为进一步开发打下基础。
以上就是使用python从零实现一个最基础的mcp协议的详细内容，更多关于python实现mcp协议的资料请关注代码网其它相关文章！
发表评论