@Slf4j
publicclassMcpClient {
    
    private String baseUri;        // MCP 服务器基础地址
    private String sseEndpoint;    // SSE 连接端点
    private String messageEndpoint; // HTTP POST 消息端点
    private SseClient sseClient = new SseClient(); // SSE 客户端实例
    
    // 异步任务管理 - 用于同步连接状态和请求响应
    private CompletableFuture<Void> connectFuture = new CompletableFuture<>();  // 连接建立完成信号
    private CompletableFuture<Void> initFuture = new CompletableFuture<>();     // 初始化完成信号
    
    // 待处理的请求映射表：requestId -> 对应的Future对象
    privatefinal Map<String, CompletableFuture<Map<String, Object>>> pendingRequests = new ConcurrentHashMap<>();
    
    privatefinal ObjectMapper objectMapper = new ObjectMapper(); // JSON 序列化工具
    
    /**
     * MCP 客户端构造函数
     * @param baseUri MCP 服务器地址
     */
    publicMcpClient(String baseUri){
        this.baseUri = baseUri;
        this.sseEndpoint = baseUri;
        
        // MCP 协议规定的三步初始化流程
        this.connect();           // 1. 建立 SSE 连接，获取 POST 端点
        this.sendInitialize();    // 2. 发送初始化请求，协商协议版本和能力
        this.sendNotification("notifications/initialized", null);  // 3. 发送初始化完成通知
    }
}

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

field: value\n\n          /* 每条消息以两个换行结束 */

event: endpoint
data: /message?sessionId=fc92c0fc-6222-466c-b353-bdc4931980a6

event: message
data: {"jsonrpc":"2.0","id":"0","result":{"capabilities":{"tools":{"listChanged":true}},
"serverInfo":{"name":"SpringBoot MCP Server","version":"1.0.0"},"protocolVersion":"2024-11-05"}}

event: message
data: {"jsonrpc":"2.0","id":"1","result":{"tools":[{"inputSchema":{"type":"object","properties":
{"name":{"description":"Name to greet (optional)","type":"string"}}},"name":"hello_world","description":"Returns a Hello World message"},
{"inputSchema":{"type":"object","properties":{}},"name":"get_time","description":"Returns current server time"},
{"de-snippet__string">"inputSchema":type":"object","properties":{"message":{"description":"Message to echo back","type":"string"}}},
"name":"echo","description":"Echoes back the provided message"}]}}

event: message
data: {"jsonrpc":"2.0","id":"2","result":{"content":[{"text":"Hello, 宸游!","type":"text"}]}}

@GetMapping(value="/stock-price", produces=TEXT_EVENT_STREAM_VALUE)
public SseEmitter stream(){
    SseEmitter e = new SseEmitter();
    new Thread(() -> {
        while(true){
            e.send(SseEmitter.event().data(price));
            sleep(1000);
        }
    }).start();
    return e;
}

const es = new EventSource("/stock-price");
es.onmessage = e => document.body.innerText = e.data;

@GetMapping("/stream-sse")
public Flux<ServerSentEvent<String>> stream() {
    return Flux.interval(Duration.ofSeconds(1))
               .map(i -> ServerSentEvent.builder()
                       .id(i.toString())
                       .event("tick")
                       .data("now=" + LocalTime.now())
                       .build());
}

{ "jsonrpc": "2.0", "method": "...", "params": {...}, "id": <unique> }

{ "jsonrpc": "2.0", "id": <same>, "result": {...} }   // 成功
{ "jsonrpc": "2.0", "id": <same>, "error": {code, message, data?} } // 失败

{
  "jsonrpc": "2.0",
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "sampling": {},
      "roots": { "listChanged": true }
    },
    "clientInfo": { "name": "mcp-inspector", "version": "0.9.0" }
  },
  "id": 0
}

springboot-mcp-server/
├── pom.xml
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   └── com/
│   │   │       └── example/
│   │   │           ├── HelloWorldApplication.java
│   │   │           ├── config/ --http配置
│   │   │           │   └── McpConfig.java
│   │   │           └── mcp/
│   │   │               ├── protocol/ --jsonrpc2.0格式的请求和响应
│   │   │               │   ├── McpRequest.java
│   │   │               │   └── McpResponse.java
│   │   │               ├── transport/ -- sse与http post传输端点
│   │   │               │   └── SseTransport.java 
│   │   │               ├── server/ -- mcp协议处理
│   │   │               │   └── McpServerHandler.java
│   │   │               └── tools/ -- 工具注册、调用
│   │   │                   └── McpToolRegistry.java
│   │   └── resources/
│   │       ├── application.properties
│   │       └── static/
│   │           └── mcp-test.html
│   └── test/
│       └── java/
│           └── com/
│               └── example/
│                   └── HelloWorldApplicationTests.java
└── README.md

/**
 * 标准 MCP SSE 端点 - 符合 MCP 协议规范
 * 客户端连接此端点接收服务器消息
 */
@GetMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> sseEndpoint(@RequestParam(required = false) String clientId,
                                                 ServerHttpRequest request) {
    // 安全检查：验证 Origin 头防止 DNS 重绑定攻击
    String origin = request.getHeaders().getFirst("Origin");
    if (origin != null && !isValidOrigin(origin)) {
        System.err.println("Invalid origin rejected: " + origin);
        return Flux.error(new SecurityException("Invalid origin"));
    }
    
    String sessionId = clientId != null ? clientId : "client-" + System.currentTimeMillis();
    
    // 为每个客户端创建独立的消息流
    Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().multicast().onBackpressureBuffer();
    clientSinks.put(sessionId, sink);
    
    System.out.println("MCP SSE client connected: " + sessionId + " from origin: " + origin);
    
    // 创建心跳流，每30秒发送一次心跳
    Flux<ServerSentEvent<String>> heartbeat = Flux.interval(Duration.ofSeconds(30))
        .map(tick -> ServerSentEvent.<String>builder()
            .event("ping")
            .data("{\"type\":\"ping\"}")
            .build());
    
    // 合并消息流和心跳流
    return Flux.merge(
        sink.asFlux(),
        heartbeat
    )
    .doOnSubscribe(subscription -> {
        // 根据 MCP 协议，必须发送 endpoint 事件
        try {
            // 1. 发送 endpoint 事件 - MCP 协议要求
            String endpointUri = getBaseUrl(request) + "/message/" + sessionId;
            
            // 直接发送 URI 字符串，不包装在对象中
            ServerSentEvent<String> endpointEvent = ServerSentEvent.<String>builder()
                .event("endpoint")
                .data(endpointUri)
                .build();
            
            sink.tryEmitNext(endpointEvent);
            System.out.println("Sent endpoint event to client: " + sessionId + " with URI: " + endpointUri);
            
        } catch (Exception e) {
            System.err.println("Error sending endpoint event: " + e.getMessage());
        }
    })
    .doOnCancel(() -> {
        System.out.println("MCP SSE client disconnected: " + sessionId);
        clientSinks.remove(sessionId);
        sink.tryEmitComplete();
    })
    .doOnError(error -> {
        System.err.println("MCP SSE error for client " + sessionId + ": " + error.getMessage());
        clientSinks.remove(sessionId);
    })
    .onErrorResume(error -> {
        System.err.println("SSE stream error, attempting to recover: " + error.getMessage());
        return Flux.empty();
    });
}

/**
 * 处理客户端发送的 MCP 请求 - 支持会话特定端点
 * 客户端通过 POST 请求发送 MCP 消息到指定的端点
 */
@PostMapping(value = "/message/{sessionId}", consumes = MediaType.APPLICATION_JSON_VALUE)
public Mono<Void> handleSessionMessage(@PathVariable String sessionId, 
                                      @RequestBody String messageJson,
                                      ServerHttpRequest request) {
    return handleMessageInternal(sessionId, messageJson, request);
}


private Mono<Void> handleMessageInternal(String sessionId, String messageJson, ServerHttpRequest request){
    return Mono.fromRunnable(() -> {
        try {
            // 安全检查：验证 Origin 头
            String origin = request.getHeaders().getFirst("Origin");
            if (origin != null && !isValidOrigin(origin)) {
                System.err.println("Invalid origin rejected for message: " + origin);
                return;
            }
            
            // 解析消息
            McpRequest mcpRequest = objectMapper.readValue(messageJson, McpRequest.class);
            System.out.println("Received MCP request: " + mcpRequest.getMethod() + 
                             " (id: " + mcpRequest.getId() + ") from session: " + sessionId);
            
            // 处理请求
            McpResponse response = serverHandler.handleRequest(mcpRequest);
            
            // 通过 SSE 发送响应
            if (response != null) {
                sendMessageToClient(sessionId, response);
            }
            
        } catch (Exception e) {
            System.err.println("Error processing MCP message: " + e.getMessage());
            e.printStackTrace();
            
            // 发送错误响应
            try {
                McpRequest mcpReq = objectMapper.readValue(messageJson, McpRequest.class);
                McpResponse errorResponse = new McpResponse();
                errorResponse.setId(mcpReq.getId());
                errorResponse.setError(new McpResponse.McpError(-32603, "Internal error: " + e.getMessage()));
                sendMessageToClient(sessionId, errorResponse);
            } catch (Exception ex) {
                System.err.println("Error sending error response: " + ex.getMessage());
            }
        }
    });
}


/**
 * 向指定客户端发送 MCP 消息
 */
publicvoidsendMessageToClient(String sessionId, Object message){
    Sinks.Many<ServerSentEvent<String>> sink = clientSinks.get(sessionId);
    if (sink != null) {
        try {
            String jsonData = objectMapper.writeValueAsString(message);
            ServerSentEvent<String> event = ServerSentEvent.<String>builder()
                .event("message")
                .data(jsonData)
                .build();
            
            sink.tryEmitNext(event);
            System.out.println("Sent MCP message to client: " + sessionId);
        } catch (Exception e) {
            System.err.println("Error sending message to client " + sessionId + ": " + e.getMessage());
        }
    } else {
        System.err.println("No active connection found for session: " + sessionId);
    }
}

public McpResponse handleRequest(McpRequest request){
    String method = request.getMethod();
    String id = request.getId();
    Map<String, Object> params = request.getParams();
    
    try {
        switch (method) {
            case"initialize":
                return handleInitialize(id, params);
                
            case"tools/list":
                return handleListTools(id);
                
            case"tools/call":
                return handleCallTool(id, params);
                
            case"notifications/initialized":
                // 客户端初始化完成通知，不需要响应
                this.initialized = true;
                System.out.println("Client initialization completed");
                return null;
                
            default:
                // 创建错误响应，确保只包含 error 字段，不包含 result
                McpResponse errorResponse = new McpResponse();
                errorResponse.setId(id);
                errorResponse.setError(new McpResponse.McpError(-32601, "Method not found: " + method));
                return errorResponse;
        }
    } catch (Exception e) {
        System.err.println("Error handling request " + method + ": " + e.getMessage());
        // 创建错误响应，确保只包含 error 字段，不包含 result
        McpResponse errorResponse = new McpResponse();
        errorResponse.setId(id);
        errorResponse.setError(new McpResponse.McpError(-32603, "Internal error: " + e.getMessage()));
        return errorResponse;
    }
}