java 实现 mcp sse demo

在使用 Java 实现 MCP SSE（Server-Sent Events）示例时，可以基于 Spring Boot 框架来构建一个支持 SSE 的服务器端点。SSE 是一种允许服务器向客户端推送实时更新的技术，非常适合用于 MCP 协议中需要保持长连接并接收服务器通知的场景。 ### 服务器端实现 以下是一个基于 Spring WebFlux 和 `SseEmitter` 的简单 MCP SSE 服务端实现示例： ```java import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.io.IOException; import java.util.concurrent.CopyOnWriteArrayList; @RestController @RequestMapping("/sse") public class SseController { private final List<SseEmitter> emitters = new CopyOnWriteArrayList<>(); @GetMapping public SseEmitter createConnection() { SseEmitter emitter = new SseEmitter(60_000L); // 设置超时时间为60秒 emitters.add(emitter); emitter.onCompletion(() -> emitters.remove(emitter)); return emitter; } public void sendEvent(String eventName, String data) { emitters.forEach(emitter -> { try { emitter.sendEvent(SseEmitter.event() .name(eventName) .data(data)); } catch (IOException e) { emitter.complete(); emitters.remove(emitter); } }); } } ``` ### 定时发送 Ping 包 为了保持连接活跃，通常会定时发送 ping 包，这可以通过 Spring 的定时任务功能实现： ```java import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; @Component public class SseScheduler { private final SseController sseController; public SseScheduler(SseController sseController) { this.sseController = sseController; } @Scheduled(fixedRate = 15_000) // 每15秒发送一次ping public void sendPing() { sseController.sendEvent("ping", "keep-alive"); } } ``` ### 客户端连接示例 Java 客户端可以使用 `EventSource` 库（如 [EventStream HTTP Client](https://github.com/AsyncHttpClient/async-http-client)）来监听服务器事件： ```java import org.asynchttpclient.DefaultAsyncHttpClient; import org.asynchttpclient.ws.WebSocket; import org.asynchttpclient.ws.WebSocketTextListener; import org.asynchttpclient.ws.listener.AbstractWebSocketListener; import java.net.URI; public class SseClient { public static void main(String[] args) throws Exception { var client = new DefaultAsyncHttpClient(); URI uri = URI.create("http://localhost:8080/sse"); WebSocket socket = client.prepareGet(uri.toString()) .execute(new AbstractWebSocketListener() { @Override public void onTextFrame(String payload, boolean finalFragment, int rsv) { System.out.println("Received event: " + payload); } @Override public void onOpen(WebSocket websocket) { System.out.println("Connection opened"); } @Override public void onClose(WebSocket websocket, int code, String reason) { System.out.println("Connection closed: " + reason); } }) .get(); Thread.sleep(60000); // 保持连接1分钟 socket.close(); client.close(); } } ``` ### 数据格式与事件流控制 MCP SSE 通信中，服务器通常以如下格式发送事件： ``` event: event-name data: event-data ``` 在 Java 中可以通过 `SseEmitter` 的 API 来构造这种格式的数据流，并确保使用 `text/event-stream` MIME 类型响应客户端请求[^3]。 ---