实现 MCP 下的 stdio/SSE 客户端与服务器

目录

• • 实现 MCP 下的 stdio/SSE 客户端与服务器
  • • 1. MCP 协议基础
    • 2. 实现 MCP 服务器
    • • 2.1 定义工具接口
      • 2.2 JSON-RPC 消息处理核心
      • 2.3 stdio 服务器实现
      • 2.4 SSE 服务器实现（Spring Boot）
    • 3. 实现 MCP 客户端
    • • 3.1 客户端接口抽象
      • 3.2 stdio 客户端实现
      • 3.3 SSE 客户端实现
    • 4. 与 Spring AI 集成
    • • 4.1 将 MCP 工具包装为 Spring AI FunctionCallback
      • 4.2 动态发现 MCP 工具
      • 4.3 配置 ChatClient
    • 5. 测试与运行
    • • 启动 MCP 服务器（stdio）
      • 启动 Spring Boot 应用（集成 MCP 客户端）
    • 6. 总结

实现 MCP 下的 stdio/SSE 客户端与服务器

Model Context Protocol（MCP）是 Anthropic 提出的开放协议，旨在标准化大语言模型与外部工具、数据源的交互。它支持两种传输方式：

• stdio：通过标准输入输出进行通信，适合本地进程间调用（如 CLI 工具）。
• SSE（Server-Sent Events）：基于 HTTP 的流式传输，适合远程服务。

本文将详细介绍如何使用 Java 实现 MCP 的客户端和服务器，支持 stdio 和 SSE，并展示如何将其集成到 Spring AI 中，让大模型能够调用 MCP 工具。

1. MCP 协议基础

MCP 基于 JSON-RPC 2.0 消息格式，主要包含以下方法：

• initialize：客户端和服务器握手，协商协议版本和能力。
• tools/list：列出服务器提供的工具。
• tools/call：调用指定工具，传入参数，返回结果。
• notifications：服务器主动推送的消息（如进度更新）。

消息结构示例：

{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "get_weather",
"arguments": {
"location": "北京"
}
}
}

响应：

{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "北京今天晴，25℃"
}
]
}
}

2. 实现 MCP 服务器

我们将分别实现 stdio 和 SSE 两种传输的服务器，并共享相同的工具注册和调用逻辑。

2.1 定义工具接口

首先定义一个通用的工具处理器：

public interface McpTool {
String getName();
String getDescription();
JsonSchema getInputSchema(); // 可以用 JSON Schema 对象
Object execute(Map<String, Object> arguments);
}

简单实现一个天气工具：

@Component
public class WeatherTool implements McpTool {
@Override
public String getName() { return "get_weather"; }
@Override
public String getDescription() { return "获取指定城市的天气"; }
@Override
public JsonSchema getInputSchema() {
return JsonSchema.builder()
.addProperty("location", JsonSchema.string().description("城市名称"))
.build();
}
@Override
public Object execute(Map<String, Object> arguments) {
String location = (String) arguments.get("location");
// 实际调用天气API
return Map.of("content", List.of(Map.of("type", "text", "text", location + "天气晴")));
}
}

2.2 JSON-RPC 消息处理核心

实现一个核心处理器，解析 JSON-RPC 消息并分发到对应方法：

public class McpRequestHandler {
private final Map<String, McpTool> tools;

public McpRequestHandler(List<McpTool> tools) {
this.tools = tools.stream().collect(Collectors.toMap(McpTool::getName, t -> t));
}

public String handleRequest(String requestJson) {
JsonNode req = new ObjectMapper().readTree(requestJson);
String method = req.get("method").asText();
JsonNode params = req.get("params");
long id = req.get("id").asLong();

try {
Object result;
switch (method) {
case "initialize":
result = handleInitialize(params);
break;
case "tools/list":
result = handleToolsList();
break;
case "tools/call":
result = handleToolsCall(params);
break;
default:
throw new McpException("未知方法: " + method);
}
return new ObjectMapper().writeValueAsString(Map.of(
"jsonrpc", "2.0",
"id", id,
"result", result
));
} catch (Exception e) {
return new ObjectMapper().writeValueAsString(Map.of(
"jsonrpc", "2.0",
"id", id,
"error", Map.of("code", –32000, "message", e.getMessage())
));
}
}

private Map<String, Object> handleInitialize(JsonNode params) {
return Map.of("protocolVersion", "0.1.0", "capabilities", Map.of());
}

private List<Map<String, Object>> handleToolsList() {
return tools.values().stream()
.map(t -> Map.of(
"name", t.getName(),
"description", t.getDescription(),
"inputSchema", t.getInputSchema()
))
.collect(Collectors.toList());
}

private Object handleToolsCall(JsonNode params) {
String name = params.get("name").asText();
JsonNode arguments = params.get("arguments");
McpTool tool = tools.get(name);
if (tool == null) throw new McpException("工具不存在: " + name);
Map<String, Object> argsMap = new ObjectMapper().convertValue(arguments, new TypeReference<>() {});
return tool.execute(argsMap);
}
}

2.3 stdio 服务器实现

stdio 服务器从标准输入读取请求，处理后写入标准输出。注意使用 System.in 和 System.out，并正确处理流。

public class StdioMcpServer {
private final McpRequestHandler handler;

public StdioMcpServer(List<McpTool> tools) {
this.handler = new McpRequestHandler(tools);
}

public void start() throws IOException {
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(System.out));

String line;
while ((line = reader.readLine()) != null) {
if (line.trim().isEmpty()) continue;
String response = handler.handleRequest(line);
writer.write(response);
writer.newLine();
writer.flush();
}
}

public static void main(String[] args) {
List<McpTool> tools = List.of(new WeatherTool());
new StdioMcpServer(tools).start();
}
}

注意事项：

• MCP 通常要求每条消息以换行符分隔。
• 错误处理要完善，避免程序崩溃。

2.4 SSE 服务器实现（Spring Boot）

使用 Spring Web MVC 的 SseEmitter 实现 SSE 传输。每个客户端连接对应一个 SseEmitter，我们需要维护连接，并通过 HTTP POST 接收客户端发来的 JSON-RPC 请求，然后将响应通过 SSE 流发送回去。

MCP over SSE 的工作方式：

• 客户端先通过 HTTP GET 建立一个 SSE 连接，服务器返回 endpoint 事件，告知客户端后续请求的 POST URL。
• 客户端通过 POST 发送 JSON-RPC 请求到指定 endpoint，服务器处理后通过 SSE 连接返回响应。

因此，我们需要两个端点：

@RestController
public class McpSseController {
private final McpRequestHandler handler;
private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

public McpSseController(List<McpTool> tools) {
this.handler = new McpRequestHandler(tools);
}

@GetMapping("/sse")
public SseEmitter handleSse() {
SseEmitter emitter = new SseEmitter(0L); // 永不超时
String sessionId = UUID.randomUUID().toString();
emitters.put(sessionId, emitter);

// 发送 endpoint 事件，告知客户端消息 POST 地址
try {
emitter.send(SseEmitter.event()
.name("endpoint")
.data("/messages?sessionId=" + sessionId));
} catch (IOException e) {
emitter.completeWithError(e);
}

emitter.onCompletion(() -> emitters.remove(sessionId));
emitter.onTimeout(() -> emitters.remove(sessionId));
return emitter;
}

@PostMapping("/messages")
public void receiveMessage(@RequestParam String sessionId, @RequestBody String requestJson) {
SseEmitter emitter = emitters.get(sessionId);
if (emitter == null) {
throw new ResponseStatusException(HttpStatus.NOT_FOUND, "Session not found");
}
// 处理请求，并通过 SSE 发送响应
String responseJson = handler.handleRequest(requestJson);
try {
emitter.send(SseEmitter.event()
.name("message")
.data(responseJson));
} catch (IOException e) {
emitter.completeWithError(e);
}
}
}

注意：

• 每个会话对应一个 SSE 连接，服务器通过 sessionId 关联。
• 实际生产环境需要处理心跳和断线重连。

3. 实现 MCP 客户端

客户端需要支持 stdio 和 SSE 两种方式连接到服务器，并发送 JSON-RPC 请求，接收响应。

3.1 客户端接口抽象

定义一个通用的 McpClient 接口：

public interface McpClient {
CompletableFuture<JsonNode> call(String method, JsonNode params);
void close();
}

3.2 stdio 客户端实现

stdio 客户端启动一个子进程（即 stdio 服务器），通过进程的输入输出流通信。

public class StdioMcpClient implements McpClient {
private final Process process;
private final BufferedWriter writer;
private final BufferedReader reader;
private final Map<Long, CompletableFuture<JsonNode>> pending = new ConcurrentHashMap<>();
private long nextId = 1;
private final Thread readerThread;

public StdioMcpClient(String command) throws IOException {
ProcessBuilder pb = new ProcessBuilder(command.split(" "));
process = pb.start();
writer = new BufferedWriter(new OutputStreamWriter(process.getOutputStream()));
reader = new BufferedReader(new InputStreamReader(process.getInputStream()));

readerThread = new Thread(this::readLoop);
readerThread.setDaemon(true);
readerThread.start();

// 发送初始化请求（可选）
initialize();
}

private void readLoop() {
try {
String line;
while ((line = reader.readLine()) != null) {
JsonNode response = new ObjectMapper().readTree(line);
long id = response.get("id").asLong();
CompletableFuture<JsonNode> future = pending.remove(id);
if (future != null) {
future.complete(response);
}
}
} catch (IOException e) {
// 处理异常
}
}

@Override
public CompletableFuture<JsonNode> call(String method, JsonNode params) {
long id = nextId++;
Map<String, Object> request = Map.of(
"jsonrpc", "2.0",
"id", id,
"method", method,
"params", params
);
CompletableFuture<JsonNode> future = new CompletableFuture<>();
pending.put(id, future);
try {
writer.write(new ObjectMapper().writeValueAsString(request));
writer.newLine();
writer.flush();
} catch (IOException e) {
future.completeExceptionally(e);
}
return future;
}

private void initialize() {
// 发送初始化请求，忽略结果
call("initialize", JsonNodeFactory.instance.objectNode());
}

@Override
public void close() {
process.destroy();
readerThread.interrupt();
}
}

3.3 SSE 客户端实现

SSE 客户端需要先建立 SSE 连接，获取消息 POST 地址，然后通过 HTTP POST 发送请求，同时监听 SSE 事件接收响应。

public class SseMcpClient implements McpClient {
private final String baseUrl;
private final WebClient webClient;
private final Map<Long, CompletableFuture<JsonNode>> pending = new ConcurrentHashMap<>();
private long nextId = 1;
private volatile String messageUrl;

public SseMcpClient(String baseUrl) {
this.baseUrl = baseUrl;
this.webClient = WebClient.builder().baseUrl(baseUrl).build();
connectSse();
}

private void connectSse() {
webClient.get().uri("/sse")
.accept(MediaType.TEXT_EVENT_STREAM)
.exchangeToFlux(response -> response.bodyToFlux(ServerSentEvent.class))
.doOnNext(event -> {
if ("endpoint".equals(event.event())) {
this.messageUrl = baseUrl + event.data();
} else if ("message".equals(event.event())) {
handleResponse(event.data());
}
})
.subscribe();
}

private void handleResponse(String data) {
try {
JsonNode response = new ObjectMapper().readTree(data);
long id = response.get("id").asLong();
CompletableFuture<JsonNode> future = pending.remove(id);
if (future != null) {
future.complete(response);
}
} catch (IOException e) {
// 记录错误
}
}

@Override
public CompletableFuture<JsonNode> call(String method, JsonNode params) {
long id = nextId++;
Map<String, Object> request = Map.of(
"jsonrpc", "2.0",
"id", id,
"method", method,
"params", params
);
CompletableFuture<JsonNode> future = new CompletableFuture<>();
pending.put(id, future);

// 通过 POST 发送请求到 messageUrl
webClient.post().uri(messageUrl)
.contentType(MediaType.APPLICATION_JSON)
.bodyValue(request)
.retrieve()
.bodyToMono(Void.class)
.subscribe(); // 忽略 POST 响应，响应会通过 SSE 返回

return future;
}

@Override
public void close() {
// 关闭连接（WebClient 会自动释放）
}
}

4. 与 Spring AI 集成

现在我们已经有了 MCP 客户端，可以将其作为工具提供给 Spring AI 的 ChatClient，让大模型能够调用 MCP 服务器上的工具。

4.1 将 MCP 工具包装为 Spring AI FunctionCallback

@Component
public class McpToolCallback implements FunctionCallback {
private final McpClient mcpClient;
private final String toolName;

public McpToolCallback(McpClient mcpClient, String toolName) {
this.mcpClient = mcpClient;
this.toolName = toolName;
}

@Override
public String getName() { return toolName; }

@Override
public String getDescription() {
// 可以从 MCP 服务器获取描述，但这里简化
return "通过 MCP 调用的工具: " + toolName;
}

@Override
public String getInputTypeSchema() {
// 可以从 MCP 服务器获取 schema
return "{\\"type\\":\\"object\\",\\"properties\\":{}}";
}

@Override
public Object apply(Map<String, Object> arguments) {
// 调用 MCP 服务器的 tools/call
JsonNode params = new ObjectMapper().valueToTree(Map.of(
"name", toolName,
"arguments", arguments
));
JsonNode response = mcpClient.call("tools/call", params).join(); // 同步等待，实际可异步
return response.get("result").get("content").get(0).get("text").asText();
}
}

4.2 动态发现 MCP 工具

我们可以先通过 MCP 的 tools/list 获取所有可用工具，然后为每个工具创建一个 FunctionCallback 并注册到 Spring AI。

@Service
public class McpToolDiscovery {
private final McpClient mcpClient;

public McpToolDiscovery(McpClient mcpClient) {
this.mcpClient = mcpClient;
}

public List<FunctionCallback> discoverTools() {
JsonNode result = mcpClient.call("tools/list", JsonNodeFactory.instance.objectNode()).join();
List<FunctionCallback> callbacks = new ArrayList<>();
for (JsonNode tool : result.get("result")) {
String name = tool.get("name").asText();
callbacks.add(new McpToolCallback(mcpClient, name));
}
return callbacks;
}
}

4.3 配置 ChatClient

在 Spring Boot 配置中，创建 ChatClient 并注册这些动态发现的工具。

@Configuration
public class AiConfig {

@Bean
public ChatClient chatClient(ChatModel chatModel, McpToolDiscovery discovery) {
List<FunctionCallback> tools = discovery.discoverTools();
return ChatClient.builder(chatModel)
.defaultTools(tools.toArray(new FunctionCallback[0]))
.build();
}
}

5. 测试与运行

启动 MCP 服务器（stdio）

java -cp your-app.jar com.example.StdioMcpServer

启动 Spring Boot 应用（集成 MCP 客户端）

配置文件中指定使用哪种客户端：

mcp:
type: stdio
command: "java -cp your-app.jar com.example.StdioMcpServer" # 如果服务器独立进程
# 或者 SSE
# type: sse
# url: http://localhost:8080

然后注入 ChatClient 即可在对话中调用这些工具。

6. 总结

本文详细介绍了如何使用 Java 实现 MCP 协议的 stdio 和 SSE 客户端与服务器，并展示了如何将其集成到 Spring AI 中，使大模型能够调用远程工具。通过 MCP，我们可以将各种现有服务（数据库、API、本地工具）统一暴露给 LLM，构建强大的智能体应用。

随着 MCP 生态的发展，未来 Spring AI 可能会提供官方集成，但目前通过自定义实现，我们已经可以充分利用这一协议。