深入解析MCP实现方式
对于模型 (Gemini/GPT) 来说,交互是完全一样的!模型只负责在看到 “Tool Definitions”(工具定义) 和 “User Request”(用户请求) 后,生成 “Tool Call”(工具调用) JSON。所有的区别都在于编排器 (MCP 客户端):当编排器看到command它知道必须启动一个子进程,并通过stdinstdout与之通信。(如我们第一个答案所示)当编排器看到
mcpServers 的 command + args 模式(我们称之为 STDIO 模式)只是实现 MCP(模型上下文协议)的一种方式。
最主要的替代方案是 HTTP 模式,此外还有 IPC Socket 模式。
它们的核心区别在于通信渠道和生命周期管理。
核心区别对比
| 特性 | STDIN/STDOUT (标准输入/输出) | HTTP (网络服务) | IPC Socket (Unix 域套接字) |
|---|---|---|---|
| 通信渠道 | 进程的标准输入 (stdin) 和标准输出 (stdout) |
TCP/IP 端口 (例如 localhost:8080) |
文件系统上的套接字文件 (例如 /tmp/mcp.sock) |
| 生命周期 | 由客户端(编排器)管理。每次需要时启动,或随客户端启动。 | 独立管理。必须作为服务预先运行。 | 独立管理。必须作为服务预先运行。 |
| 1:1 vs 1:N | 严格的 1:1(一个客户端对应一个工具进程) | 1:N(一个服务器可以被多个客户端共享) | 1:N(在同一台机器上) |
| 调试难度 | 困难。必须模拟进程并写入 stdin。 |
简单。可以使用 curl, Postman 等工具直接测试。 |
中等。需要 socat 等特定工具。 |
| 网络能力 | 否(仅限本机) | 是(工具可以运行在另一台机器上) | 否(仅限本机) |
| 配置示例 | {"command": "python", "args": [...]} |
{"url": "http://localhost:8080/rpc"} |
{"socket_path": "/tmp/mcp.sock"} |
模式一:HTTP 服务器(最常见的替代方案)
在这种模式下,MCP 服务端是一个标准的 Web 服务器。**MCP 客户端(编排器)**则是一个 HTTP 客户端。
mcpServers.json 的配置会变成:
"playwright": {
"url": "http://localhost:3001/mcp"
}
编排器看到 url,就知道它不需要启动任何进程,而是应该向这个 URL 发送一个 HTTP POST 请求。
Python 实现 (使用 Flask)
1. MCP 服务端 (http_server.py)
这是一个需要你提前手动运行的 Flask Web 服务器。
#!/usr/bin/env python
from flask import Flask, request, jsonify
import sys
import json
import os
# --- 核心逻辑 (与上一版本完全相同) ---
def log(message):
print(f"[Server Log] {message}", file=sys.stderr, flush=True)
def handle_request(request):
if request.get("method") == "write_file":
try:
params = request.get("params", {})
path = params.get("path")
content = params.get("content")
if not path or content is None:
raise ValueError("缺少 'path' 或 'content' 参数")
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, 'w', encoding='utf-8') as f:
f.write(content)
log(f"文件已写入: {path}")
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {"status": "success", "path": path}
}
except Exception as e:
return {"jsonrpc": "2.0", "id": request.get("id"), "error": {"code": -1, "message": str(e)}}
else:
return {"jsonrpc": "2.0", "id": request.get("id"), "error": {"code": -2, "message": "不支持的方法"}}
# --- 核心逻辑结束 ---
# 创建 Flask app
app = Flask(__name__)
# 这是 MCP 的入口点
@app.route('/rpc', methods=['POST'])
def rpc_entrypoint():
log("收到 HTTP 请求...")
# 1. 从 HTTP Body 获取 JSON 请求
try:
request_data = request.get_json()
if not request_data:
raise ValueError("没有 JSON body")
log(f"解析后的请求: {request_data}")
except Exception as e:
log(f"JSON 解析错误: {e}")
return jsonify({"jsonrpc": "2.0", "error": {"code": -3, "message": "无效的 JSON"}}), 400
# 2. 处理请求 (逻辑复用)
response_data = handle_request(request_data)
# 3. 将响应作为 JSON 返回
log(f"发送响应: {response_data}")
return jsonify(response_data)
if __name__ == '__main__':
log("HTTP MCP 服务端正在 0.0.0.0:8080 上启动...")
# 运行服务器。在生产环境中,你会使用 gunicorn/uvicorn。
app.run(host='0.0.0.0', port=8080)
如何运行服务端: python http_server.py (它会一直运行)
2. MCP 客户端 (http_client.py)
这个 “编排器” 现在使用 requests 库,而不是 subprocess。
import requests
import json
# 1. 模拟的模型工具调用 (不变)
model_tool_call = {
"tool_name": "file_writer",
"parameters": {
"path": "./test_output/demo_http.txt",
"content": "Hello from the HTTP MCP server!"
}
}
# 2. 转换为 MCP (JSON-RPC) 请求 (不变)
mcp_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "write_file",
"params": model_tool_call["parameters"]
}
# MCP 服务器的 URL (来自配置)
SERVER_URL = "http://localhost:8080/rpc"
print(f"[Client] 准备向 {SERVER_URL} 发送 HTTP POST 请求...")
try:
# 3. (客户端) 发送 HTTP POST 请求
# 注意:不再启动子进程,而是发送网络请求
http_response = requests.post(SERVER_URL, json=mcp_request, timeout=5)
# 检查 HTTP 状态码
http_response.raise_for_status()
# 4. (客户端) 从 HTTP 响应中获取 JSON
response_data = http_response.json()
print(f"[Client] <- 收到 HTTP 响应: {response_data}")
# 5. (编排器) 转换为 "工具结果" (不变)
if "result" in response_data:
model_tool_result = {"status": "success", "content": response_data["result"]}
else:
model_tool_result = {"status": "error", "content": response_data.get("error")}
print(f"[Client] 准备发回给模型的工具结果: {model_tool_result}")
except requests.exceptions.ConnectionError:
print(f"[Client] 错误: 无法连接到 {SERVER_URL}。服务端运行了吗?")
except requests.exceptions.RequestException as e:
print(f"[Client] HTTP 请求错误: {e}")
如何运行客户端: python http_client.py (在服务端已运行的情况下)
PHP 实现 (使用内置服务器)
1. MCP 服务端 (http_server.php 和 router.php)
PHP 的内置服务器需要一个 “路由器” 脚本。
router.php (核心逻辑):
<?php
// (与上一版本相同的 `log_msg` 和 `handle_request` 函数)
function log_msg($message) {
fwrite(STDERR, "[Server Log] " . $message . PHP_EOL);
fflush(STDERR);
}
function handle_request($request) {
// ... (省略与上一版本完全相同的代码) ...
if (isset($request['method']) && $request['method'] == 'write_file') {
try {
$params = $request['params'] ?? [];
$path = $params['path'] ?? null;
$content = $params['content'] ?? null;
if (!$path || $content === null) { throw new Exception("缺少 'path' 或 'content' 参数"); }
$dir = dirname($path);
if (!is_dir($dir)) { mkdir($dir, 0777, true); }
file_put_contents($path, $content);
log_msg("文件已写入: $path");
return ["jsonrpc" => "2.0", "id" => $request['id'] ?? null, "result" => ["status" => "success", "path" => $path]];
} catch (Exception $e) {
return ["jsonrpc" => "2.0", "id" => $request['id'] ?? null, "error" => ["code" => -1, "message" => $e->getMessage()]];
}
}
// ... (省略) ...
}
// --- HTTP 服务端逻辑 ---
log_msg("收到 HTTP 请求: " . $_SERVER['REQUEST_URI']);
// 我们只关心 /rpc 上的 POST 请求
if ($_SERVER['REQUEST_URI'] == '/rpc' && $_SERVER['REQUEST_METHOD'] == 'POST') {
// 1. 从 HTTP Body 读取原始 JSON
$raw_body = file_get_contents('php://input');
log_msg("原始请求 body: " . $raw_body);
// 2. 解析 JSON
$request_data = json_decode($raw_body, true);
if (json_last_error() !== JSON_ERROR_NONE) {
$response_data = ["jsonrpc" => "2.0", "error" => ["code" => -3, "message" => "无效的 JSON"]];
http_response_code(400);
} else {
// 3. 处理请求 (逻辑复用)
$response_data = handle_request($request_data);
}
// 4. 将响应作为 JSON 返回
header('Content-Type: application/json');
echo json_encode($response_data);
} else {
http_response_code(404);
echo json_encode(["error" => "Not Found"]);
}
如何运行服务端: php -S localhost:8080 router.php (它会一直运行)
2. MCP 客户端 (http_client.php)
这个 “编排器” 现在使用 cURL。
<?php
// 1. 模拟的模型工具调用 (不变)
$model_tool_call = [
"tool_name" => "file_writer",
"parameters" => [
"path" => "./test_output/demo_http_php.txt",
"content" => "Hello from the PHP HTTP MCP server!"
]
];
// 2. 转换为 MCP (JSON-RPC) 请求 (不变)
$mcp_request = [
"jsonrpc" => "2.0",
"id" => 2,
"method" => "write_file",
"params" => $model_tool_call["parameters"]
];
$SERVER_URL = "http://localhost:8080/rpc";
$request_json = json_encode($mcp_request);
echo "[Client] 准备向 $SERVER_URL 发送 HTTP POST 请求..." . PHP_EOL;
// 3. (客户端) 使用 cURL 发送 HTTP POST
$ch = curl_init($SERVER_URL);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
curl_setopt($ch, CURLOPT_POST, true);
curl_setopt($ch, CURLOPT_POSTFIELDS, $request_json);
curl_setopt($ch, CURLOPT_HTTPHEADER, [
'Content-Type: application/json',
'Content-Length: ' . strlen($request_json)
]);
$response_body = curl_exec($ch);
$http_code = curl_getinfo($ch, CURLINFO_HTTP_CODE);
if ($response_body === false) {
echo "[Client] cURL 错误: " . curl_error($ch) . PHP_EOL;
} else {
echo "[Client] <- 收到 HTTP 响应 (Code: $http_code): $response_body" . PHP_EOL;
// 4. (客户端) 解析 JSON 响应
$response_data = json_decode($response_body, true);
// 5. (编排器) 转换为 "工具结果" (不变)
if (isset($response_data["result"])) {
$model_tool_result = ["status" => "success", "content" => $response_data["result"]];
} else {
$model_tool_result = ["status" => "error", "content" => $response_data["error"] ?? null];
}
echo "[Client] 准备发回给模型的工具结果: " . json_encode($model_tool_result) . PHP_EOL;
}
curl_close($ch);
如何运行客户端: php http_client.php (在服务端已运行的情况下)
模式二:IPC Socket (Unix 域套接字)
这是一种介于 STDIO 和 HTTP 之间的模式。它像 HTTP 一样需要一个持久化的服务器,但它不使用网络端口,而是在文件系统上创建一个特殊的 “套接字” 文件。这更快,且更安全(仅限本机)。
mcpServers.json 的配置会变成:
"playwright": {
"socket_path": "/tmp/mcp-playwright.sock"
}
Python 实现 (仅为演示)
1. MCP 服务端 (ipc_server.py)
import socket
import os
import json
import sys
# (复用 handle_request 和 log 函数)
# ...
SOCKET_PATH = "/tmp/mcp-demo.sock"
# 确保旧的 socket 文件被删除
if os.path.exists(SOCKET_PATH):
os.remove(SOCKET_PATH)
# 创建一个 Unix 域套接字 (AF_UNIX)
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
log(f"正在绑定到 socket 文件: {SOCKET_PATH}")
server.bind(SOCKET_PATH)
server.listen(1)
log("IPC 服务端已启动,等待连接...")
while True:
try:
conn, addr = server.accept()
log("收到连接")
# 接收数据 (假设一条消息 < 4096 字节)
data = conn.recv(4096)
if not data:
break
request_data = json.loads(data.decode('utf-8'))
log(f"收到请求: {request_data}")
# 处理请求
response_data = handle_request(request_data)
# 发送响应
conn.sendall(json.dumps(response_data).encode('utf-8'))
except Exception as e:
log(f"处理连接时出错: {e}")
finally:
conn.close()
如何运行服务端: python ipc_server.py
2. MCP 客户端 (ipc_client.py)
import socket
import json
# (复用 model_tool_call 和 mcp_request)
# ...
mcp_request = { "id": 3, "method": "write_file", "params": {
"path": "./test_output/demo_ipc.txt",
"content": "Hello from IPC Socket!"
}}
SOCKET_PATH = "/tmp/mcp-demo.sock"
print(f"[Client] 准备连接到 IPC Socket: {SOCKET_PATH}")
try:
# 1. (客户端) 创建一个 AF_UNIX 套接字并连接
client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
client.connect(SOCKET_PATH)
# 2. (客户端) 发送请求
request_json = json.dumps(mcp_request)
client.sendall(request_json.encode('utf-8'))
# 3. (客户端) 接收响应
response_data = client.recv(4096).decode('utf-8')
print(f"[Client] <- 收到响应: {response_data}")
# 4. (编排器) 转换为 "工具结果" ...
# ...
except socket.error as e:
print(f"[Client] Socket 错误: {e}")
finally:
client.close()
如何运行客户端: python ipc_client.py
总结:与模型的交互有何不同?
对于模型 (Gemini/GPT) 来说,交互是完全一样的!
模型只负责在看到 “Tool Definitions”(工具定义) 和 “User Request”(用户请求) 后,生成 “Tool Call”(工具调用) JSON。
所有的区别都在于编排器 (MCP 客户端):
- 当编排器看到
command配置时: 它知道必须启动一个子进程,并通过stdin/stdout与之通信。(如我们第一个答案所示) - 当编排器看到
url配置时: 它知道必须发送一个 HTTP POST 请求到该 URL。(如本答案中的 HTTP 示例所示) - 当编排器看到
socket_path配置时: 它知道必须连接到该文件套接字进行通信。(如本答案中的 IPC 示例所示)
模型是“决策者”,而编排器是“调度员”,mcpServers.json 则是调度员的“通讯录”,告诉它该用什么方法(打电话、发邮件、还是用内部管道)去联系它的工具。
火山引擎开发者社区是火山引擎打造的AI技术生态平台,聚焦Agent与大模型开发,提供豆包系列模型(图像/视频/视觉)、智能分析与会话工具,并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长,新用户可领50万Tokens权益,助力构建智能应用。
更多推荐
所有评论(0)