82.MCP-stdio在springAI中的实现在实际的AI应用开发中我们经常需要将外部AI模型或工具集成到Spring生态系统中。最近在项目中遇到了一个需求如何通过标准输入输出stdio方式与MCPModel Control Protocol服务进行通信并在SpringAI框架中实现高效集成。网上相关资料比较零散特别是针对MCP-stdio协议在SpringAI中的具体实现方案缺乏系统性的教程。本文将完整介绍MCP-stdio协议在SpringAI中的实现方案从基础概念到完整代码示例涵盖环境搭建、协议解析、服务集成等关键环节。无论你是刚开始接触SpringAI的开发者还是需要将现有MCP服务集成到Java项目中的工程师都能从本文找到实用的解决方案。1. MCP-stdio协议基础概念1.1 什么是MCP协议MCPModel Control Protocol是一种用于与AI模型进行通信的协议标准它定义了模型服务的统一接口规范。MCP协议的核心目标是提供一种标准化的方式来调用和管理各种AI模型无论这些模型是本地部署还是云端服务。MCP-stdio是MCP协议的一种实现方式它通过标准输入输出stdin/stdout进行数据交换。这种方式的优势在于跨平台兼容性好不需要复杂的网络配置特别适合本地模型服务的集成。1.2 MCP-stdio的工作原理MCP-stdio协议基于JSON-RPC 2.0规范通过标准输入输出流进行消息传递。其基本工作流程如下客户端通过stdin向服务端发送JSON-RPC请求服务端通过stdout返回JSON-RPC响应双方通过换行符分隔每条消息消息体为UTF-8编码的JSON字符串这种设计使得任何支持标准输入输出的编程语言都能实现MCP-stdio客户端或服务端大大提高了协议的通用性。1.3 SpringAI框架简介SpringAI是Spring官方推出的AI应用开发框架它提供了一套统一的API来集成各种AI模型和服务。SpringAI的核心特性包括统一的模型抽象接口自动配置和依赖注入与Spring生态系统的无缝集成支持多种AI模型提供商OpenAI、Azure、本地模型等通过SpringAI开发者可以用相似的方式调用不同的AI服务大大降低了集成复杂度。2. 环境准备与项目搭建2.1 开发环境要求在开始实现之前需要确保开发环境满足以下要求JDK 17或更高版本Maven 3.6 或 Gradle 7.xSpring Boot 3.2.0SpringAI 1.0.0一个可用的MCP-stdio服务如本地运行的AI模型2.2 创建Spring Boot项目使用Spring Initializr创建基础项目结构curl https://start.spring.io/starter.zip \ -d dependenciesweb,ai \ -d typemaven-project \ -d languagejava \ -d bootVersion3.2.0 \ -d baseDirmcp-springai-demo \ -d groupIdcom.example \ -d artifactIdmcp-springai-demo \ -o mcp-springai-demo.zip解压后得到的基础项目结构如下mcp-springai-demo/ ├── src/ │ ├── main/ │ │ ├── java/ │ │ │ └── com/example/ │ │ │ └── McpSpringaiDemoApplication.java │ │ └── resources/ │ │ └── application.properties │ └── test/ └── pom.xml2.3 添加必要依赖在pom.xml中添加SpringAI和相关依赖?xml version1.0 encodingUTF-8? project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version3.2.0/version relativePath/ /parent groupIdcom.example/groupId artifactIdmcp-springai-demo/artifactId version0.0.1-SNAPSHOT/version properties java.version17/java.version spring-ai.version1.0.0/spring-ai.version /properties dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.ai/groupId artifactIdspring-ai-core/artifactId version${spring-ai.version}/version /dependency dependency groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies /project3. MCP-stdio协议解析与实现3.1 JSON-RPC消息结构MCP-stdio基于JSON-RPC 2.0协议我们需要先定义基本的消息结构// 文件路径src/main/java/com/example/mcp/model/JsonRpcRequest.java package com.example.mcp.model; import com.fasterxml.jackson.annotation.JsonProperty; import lombok.Data; Data public class JsonRpcRequest { JsonProperty(jsonrpc) private String jsonrpc 2.0; private String id; private String method; private Object params; public JsonRpcRequest(String method, Object params) { this.method method; this.params params; this.id java.util.UUID.randomUUID().toString(); } }// 文件路径src/main/java/com/example/mcp/model/JsonRpcResponse.java package com.example.mcp.model; import com.fasterxml.jackson.annotation.JsonProperty; import lombok.Data; Data public class JsonRpcResponse { JsonProperty(jsonrpc) private String jsonrpc 2.0; private String id; private Object result; private JsonRpcError error; Data public static class JsonRpcError { private int code; private String message; private Object data; } }3.2 MCP-stdio客户端实现创建MCP-stdio客户端负责与外部MCP服务进行通信// 文件路径src/main/java/com/example/mcp/client/McpStdioClient.java package com.example.mcp.client; import com.example.mcp.model.JsonRpcRequest; import com.example.mcp.model.JsonRpcResponse; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import java.io.*; import java.util.concurrent.*; Slf4j Component public class McpStdioClient { private final ObjectMapper objectMapper new ObjectMapper(); private Process mcpProcess; private BufferedReader reader; private BufferedWriter writer; private final ExecutorService executorService Executors.newSingleThreadExecutor(); private final ConcurrentMapString, CompletableFutureJsonRpcResponse pendingRequests new ConcurrentHashMap(); public void startMcpService(String command) throws IOException { ProcessBuilder processBuilder new ProcessBuilder(command.split( )); mcpProcess processBuilder.start(); reader new BufferedReader(new InputStreamReader(mcpProcess.getInputStream())); writer new BufferedWriter(new OutputStreamWriter(mcpProcess.getOutputStream())); // 启动响应监听线程 executorService.submit(this::listenForResponses); } public CompletableFutureJsonRpcResponse sendRequest(JsonRpcRequest request) { CompletableFutureJsonRpcResponse future new CompletableFuture(); pendingRequests.put(request.getId(), future); try { String requestJson objectMapper.writeValueAsString(request); log.debug(Sending MCP request: {}, requestJson); synchronized (writer) { writer.write(requestJson); writer.newLine(); writer.flush(); } } catch (IOException e) { future.completeExceptionally(e); pendingRequests.remove(request.getId()); } return future; } private void listenForResponses() { try { String line; while ((line reader.readLine()) ! null) { log.debug(Received MCP response: {}, line); JsonRpcResponse response objectMapper.readValue(line, JsonRpcResponse.class); CompletableFutureJsonRpcResponse future pendingRequests.remove(response.getId()); if (future ! null) { future.complete(response); } } } catch (IOException e) { log.error(Error reading MCP responses, e); // 处理所有未完成的请求 pendingRequests.values().forEach(future - future.completeExceptionally(e)); pendingRequests.clear(); } } public void stop() { if (mcpProcess ! null) { mcpProcess.destroy(); } executorService.shutdown(); } }3.3 MCP服务配置类创建配置类来管理MCP服务的启动参数和连接配置// 文件路径src/main/java/com/example/mcp/config/McpConfig.java package com.example.mcp.config; import com.example.mcp.client.McpStdioClient; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import javax.annotation.PreDestroy; import java.io.IOException; Slf4j Configuration public class McpConfig { Value(${mcp.service.command:python mcp_server.py}) private String mcpServiceCommand; private McpStdioClient mcpClient; Bean public McpStdioClient mcpStdioClient() throws IOException { mcpClient new McpStdioClient(); mcpClient.startMcpService(mcpServiceCommand); log.info(MCP stdio service started with command: {}, mcpServiceCommand); return mcpClient; } PreDestroy public void cleanup() { if (mcpClient ! null) { mcpClient.stop(); log.info(MCP stdio service stopped); } } }4. SpringAI模型集成4.1 自定义ChatModel实现创建基于MCP-stdio的自定义ChatModel这是SpringAI集成的核心// 文件路径src/main/java/com/example/mcp/ai/McpChatModel.java package com.example.mcp.ai; import com.example.mcp.client.McpStdioClient; import com.example.mcp.model.JsonRpcRequest; import com.example.mcp.model.JsonRpcResponse; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.model.Generation; import org.springframework.ai.chat.prompt.ChatOptions; import org.springframework.ai.chat.prompt.Prompt; import org.springframework.ai.model.ModelOptionsUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; Component public class McpChatModel implements ChatModel { private final McpStdioClient mcpClient; private final ObjectMapper objectMapper new ObjectMapper(); Autowired public McpChatModel(McpStdioClient mcpClient) { this.mcpClient mcpClient; } Override public ChatResponse call(Prompt prompt) { try { // 构建MCP请求 MapString, Object params Map.of( messages, prompt.getInstructions(), model, mcp-model, temperature, 0.7 ); JsonRpcRequest request new JsonRpcRequest(chat.completions, params); // 发送请求并等待响应带超时 CompletableFutureJsonRpcResponse future mcpClient.sendRequest(request); JsonRpcResponse response future.get(30, TimeUnit.SECONDS); if (response.getError() ! null) { throw new RuntimeException(MCP service error: response.getError().getMessage()); } // 解析响应并转换为SpringAI格式 return parseChatResponse(response); } catch (Exception e) { throw new RuntimeException(MCP chat call failed, e); } } private ChatResponse parseChatResponse(JsonRpcResponse response) { // 这里需要根据实际的MCP服务响应格式进行解析 // 假设响应格式与OpenAI兼容 MapString, Object result (MapString, Object) response.getResult(); String content extractContent(result); Generation generation new Generation(content); return new ChatResponse(generation); } private String extractContent(MapString, Object result) { // 简化实现实际需要根据MCP服务的具体响应格式调整 return result.get(content).toString(); } Override public ChatOptions getDefaultOptions() { return new McpChatOptions(); } // 自定义ChatOptions实现 public static class McpChatOptions implements ChatOptions { private Double temperature 0.7; private Integer maxTokens 1000; Override public Double getTemperature() { return temperature; } public void setTemperature(Double temperature) { this.temperature temperature; } Override public Integer getMaxTokens() { return maxTokens; } public void setMaxTokens(Integer maxTokens) { this.maxTokens maxTokens; } Override public String getModel() { return mcp-model; } } }4.2 服务层封装创建服务层提供更友好的API接口// 文件路径src/main/java/com/example/mcp/service/McpAIService.java package com.example.mcp.service; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.prompt.Prompt; import org.springframework.ai.chat.prompt.SystemPromptTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.util.Map; Service public class McpAIService { private final McpChatModel chatModel; Autowired public McpAIService(McpChatModel chatModel) { this.chatModel chatModel; } public String generateResponse(String userMessage) { String systemPrompt 你是一个有用的AI助手。请根据用户的问题提供准确、有帮助的回答。 如果问题涉及技术内容请提供详细的解释和代码示例。 ; SystemPromptTemplate systemPromptTemplate new SystemPromptTemplate(systemPrompt); Prompt prompt new Prompt(userMessage, systemPromptTemplate.createMessage()); ChatResponse response chatModel.call(prompt); return response.getResult().getOutput().getContent(); } public ChatResponse chatWithContext(String userMessage, MapString, Object context) { // 支持带上下文的对话 String systemPrompt 你是一个有用的AI助手。当前对话上下文{context} 请根据上下文和用户的问题提供准确的回答。 ; SystemPromptTemplate systemPromptTemplate new SystemPromptTemplate(systemPrompt); Prompt prompt new Prompt(userMessage, systemPromptTemplate.createMessage(context)); return chatModel.call(prompt); } }5. 控制器层与API暴露5.1 REST控制器实现创建REST API控制器对外提供AI服务接口// 文件路径src/main/java/com/example/mcp/controller/McpAIController.java package com.example.mcp.controller; import com.example.mcp.service.McpAIService; import lombok.extern.slf4j.Slf4j; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.*; import java.util.Map; Slf4j RestController RequestMapping(/api/ai) public class McpAIController { private final McpAIService mcpAIService; Autowired public McpAIController(McpAIService mcpAIService) { this.mcpAIService mcpAIService; } PostMapping(/chat) public ChatResponse chat(RequestBody ChatRequest request) { log.info(Received chat request: {}, request.getMessage()); try { if (request.getContext() ! null !request.getContext().isEmpty()) { return mcpAIService.chatWithContext(request.getMessage(), request.getContext()); } else { String response mcpAIService.generateResponse(request.getMessage()); return new ChatResponse(response); } } catch (Exception e) { log.error(Chat request failed, e); throw new RuntimeException(AI service temporarily unavailable); } } GetMapping(/health) public HealthCheck health() { try { String testResponse mcpAIService.generateResponse(Hello); return new HealthCheck(healthy, MCP service is responding); } catch (Exception e) { return new HealthCheck(unhealthy, MCP service error: e.getMessage()); } } // 请求和响应DTO类 public static class ChatRequest { private String message; private MapString, Object context; // getters and setters public String getMessage() { return message; } public void setMessage(String message) { this.message message; } public MapString, Object getContext() { return context; } public void setContext(MapString, Object context) { this.context context; } } public static class HealthCheck { private String status; private String message; public HealthCheck(String status, String message) { this.status status; this.message message; } // getters public String getStatus() { return status; } public String getMessage() { return message; } } }5.2 配置文件配置application.properties文件# 应用配置 server.port8080 spring.application.namemcp-springai-demo # MCP服务配置 mcp.service.commandpython /path/to/your/mcp_server.py # 日志配置 logging.level.com.example.mcpDEBUG logging.level.org.springframework.aiINFO # Jackson配置 spring.jackson.serialization.indent_outputtrue6. 测试与验证6.1 单元测试创建针对MCP-stdio客户端的单元测试// 文件路径src/test/java/com/example/mcp/client/McpStdioClientTest.java package com.example.mcp.client; import com.example.mcp.model.JsonRpcRequest; import com.example.mcp.model.JsonRpcResponse; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.boot.test.context.SpringBootTest; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.*; SpringBootTest class McpStdioClientTest { private McpStdioClient client; BeforeEach void setUp() throws Exception { client new McpStdioClient(); // 启动一个简单的echo服务进行测试 client.startMcpService(python -c \import sys; [print(line.strip()) for line in sys.stdin]\); } AfterEach void tearDown() { if (client ! null) { client.stop(); } } Test void testSendRequest() throws Exception { JsonRpcRequest request new JsonRpcRequest(test.method, test params); CompletableFutureJsonRpcResponse future client.sendRequest(request); // 由于是echo服务我们期望收到相同的请求内容 JsonRpcResponse response future.get(5, TimeUnit.SECONDS); assertNotNull(response); assertEquals(2.0, response.getJsonrpc()); assertEquals(request.getId(), response.getId()); } }6.2 集成测试创建完整的集成测试// 文件路径src/test/java/com/example/mcp/controller/McpAIControllerIntegrationTest.java package com.example.mcp.controller; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.http.MediaType; import org.springframework.test.web.servlet.MockMvc; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.*; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*; SpringBootTest AutoConfigureMockMvc class McpAIControllerIntegrationTest { Autowired private MockMvc mockMvc; Test void testHealthEndpoint() throws Exception { mockMvc.perform(get(/api/ai/health)) .andExpect(status().isOk()) .andExpect(jsonPath($.status).exists()); } Test void testChatEndpoint() throws Exception { String chatRequest { message: Hello, how are you?, context: {} } ; mockMvc.perform(post(/api/ai/chat) .contentType(MediaType.APPLICATION_JSON) .content(chatRequest)) .andExpect(status().isOk()); } }7. 常见问题与解决方案7.1 连接稳定性问题问题现象MCP服务进程意外退出或连接中断解决方案// 增强的McpStdioClient添加重连机制 Component public class RobustMcpStdioClient extends McpStdioClient { private static final int MAX_RETRIES 3; private static final long RETRY_DELAY_MS 5000; Override public void startMcpService(String command) throws IOException { for (int attempt 1; attempt MAX_RETRIES; attempt) { try { super.startMcpService(command); log.info(MCP service started successfully on attempt {}, attempt); return; } catch (IOException e) { log.warn(Failed to start MCP service (attempt {}/{}): {}, attempt, MAX_RETRIES, e.getMessage()); if (attempt MAX_RETRIES) { throw e; } try { Thread.sleep(RETRY_DELAY_MS); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); throw new IOException(Interrupted during retry, ie); } } } } }7.2 消息序列化问题问题现象JSON序列化/反序列化失败解决方案// 增强的JSON处理添加错误处理和日志记录 public CompletableFutureJsonRpcResponse sendRequest(JsonRpcRequest request) { try { String requestJson objectMapper.writeValueAsString(request); log.debug(Sending MCP request: {}, requestJson); // 验证JSON格式 objectMapper.readTree(requestJson); synchronized (writer) { writer.write(requestJson); writer.newLine(); writer.flush(); } CompletableFutureJsonRpcResponse future new CompletableFuture(); pendingRequests.put(request.getId(), future); return future; } catch (Exception e) { log.error(Failed to serialize or send request, e); CompletableFutureJsonRpcResponse future new CompletableFuture(); future.completeExceptionally(e); return future; } }7.3 性能优化建议连接池管理对于高并发场景考虑实现连接池批量请求支持批量发送多个请求减少IO开销异步处理使用CompletableFuture进行非阻塞调用超时控制为每个请求设置合理的超时时间8. 生产环境最佳实践8.1 安全考虑在生产环境中使用MCP-stdio集成时需要注意以下安全事项// 安全增强的配置类 Configuration EnableConfigurationProperties(McpSecurityProperties.class) public class SecureMcpConfig { private final McpSecurityProperties securityProperties; public SecureMcpConfig(McpSecurityProperties securityProperties) { this.securityProperties securityProperties; } Bean public McpStdioClient mcpStdioClient() throws IOException { // 验证命令安全性 validateCommand(securityProperties.getCommand()); McpStdioClient client new McpStdioClient(); client.startMcpService(securityProperties.getCommand()); return client; } private void validateCommand(String command) { // 检查命令是否包含危险操作 if (command.contains(rm ) || command.contains(del ) || command.contains(format) || command.contains(shutdown)) { throw new SecurityException(Potentially dangerous MCP command detected); } } }8.2 监控与日志添加详细的监控和日志记录// 监控增强的McpChatModel Slf4j Component public class MonitoredMcpChatModel extends McpChatModel { private final MeterRegistry meterRegistry; private final Counter requestCounter; private final Timer responseTimer; public MonitoredMcpChatModel(McpStdioClient mcpClient, MeterRegistry meterRegistry) { super(mcpClient); this.meterRegistry meterRegistry; this.requestCounter Counter.builder(mcp.requests) .description(Number of MCP requests) .register(meterRegistry); this.responseTimer Timer.builder(mcp.response.time) .description(MCP response time) .register(meterRegistry); } Override public ChatResponse call(Prompt prompt) { requestCounter.increment(); return responseTimer.record(() - { try { return super.call(prompt); } catch (Exception e) { log.error(MCP request failed, e); throw e; } }); } }8.3 配置管理最佳实践环境隔离为不同环境dev、test、prod配置不同的MCP服务敏感信息使用Spring Cloud Config或Kubernetes Secrets管理敏感配置健康检查实现完善的健康检查端点优雅停机确保应用关闭时正确清理MCP进程通过本文的完整实现方案你可以在SpringAI框架中成功集成MCP-stdio服务构建稳定可靠的AI应用。这种集成方式不仅适用于本地AI模型也可以扩展到各种支持stdio协议的AI服务为企业的AI应用开发提供了灵活的技术选型方案。
