mcpServerscommand + 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 客户端):

  1. 当编排器看到 command 配置时: 它知道必须启动一个子进程,并通过 stdin/stdout 与之通信。(如我们第一个答案所示)
  2. 当编排器看到 url 配置时: 它知道必须发送一个 HTTP POST 请求到该 URL。(如本答案中的 HTTP 示例所示)
  3. 当编排器看到 socket_path 配置时: 它知道必须连接到该文件套接字进行通信。(如本答案中的 IPC 示例所示)

模型是“决策者”,而编排器是“调度员”,mcpServers.json 则是调度员的“通讯录”,告诉它该用什么方法(打电话、发邮件、还是用内部管道)去联系它的工具。

Logo

火山引擎开发者社区是火山引擎打造的AI技术生态平台,聚焦Agent与大模型开发,提供豆包系列模型(图像/视频/视觉)、智能分析与会话工具,并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长,新用户可领50万Tokens权益,助力构建智能应用。

更多推荐