feat: Support MCP 2025-03-26 protocol with streaming tools (#3054)

- 实现MCP协议2025-03-26版本支持
- 添加流式工具(Streamable Tools)架构
- 支持双传输模式：流式传输(SSE)和无状态传输
- 实现30+个Arthas工具的MCP接口封装
- 添加实时进度通知和结果流式输出
- 支持watch、monitor、trace、dashboard、profiler等流式工具

Author: Yeaury

core/src/main/java/com/taobao/arthas/core/command/CommandExecutorImpl.java

@@ -80,8 +80,7 @@
 
 import com.taobao.arthas.mcp.server.ArthasMcpBootstrap;
 import com.taobao.arthas.mcp.server.CommandExecutor;
-import com.taobao.arthas.mcp.server.protocol.server.handler.McpRequestHandler;
-import com.taobao.arthas.mcp.server.protocol.spec.McpServerTransportProvider;
+import com.taobao.arthas.mcp.server.protocol.server.handler.McpHttpRequestHandler;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.util.concurrent.DefaultThreadFactory;
@@ -130,7 +129,7 @@ public class ArthasBootstrap {
 
     private HttpApiHandler httpApiHandler;
 
-    private McpRequestHandler mcpRequestHandler;
+    private McpHttpRequestHandler mcpRequestHandler;
 
     private HttpSessionManager httpSessionManager;
     private SecurityAuthenticator securityAuthenticator;
@@ -683,7 +682,7 @@ public HttpApiHandler getHttpApiHandler() {
         return httpApiHandler;
     }
 
-    public McpRequestHandler getMcpRequestHandler() {
+    public McpHttpRequestHandler getMcpRequestHandler() {
         return mcpRequestHandler;
     }
 

core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java

@@ -5,7 +5,7 @@
 import com.taobao.arthas.common.IOUtils;
 import com.taobao.arthas.core.server.ArthasBootstrap;
 import com.taobao.arthas.core.shell.term.impl.http.api.HttpApiHandler;
-import com.taobao.arthas.mcp.server.protocol.server.handler.McpRequestHandler;
+import com.taobao.arthas.mcp.server.protocol.server.handler.McpHttpRequestHandler;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
@@ -38,7 +38,7 @@ public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequ
 
     private HttpApiHandler httpApiHandler;
 
-    private McpRequestHandler mcpRequestHandler;
+    private McpHttpRequestHandler mcpRequestHandler;
 
     public HttpRequestHandler(String wsUri) {
         this(wsUri, ArthasBootstrap.getInstance().getOutputPath());
@@ -76,9 +76,8 @@ protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request)
                 }
 
                 //handle mcp request
-                String messageEndpoint = mcpRequestHandler.getMessageEndpoint();
-                String sseEndpoint = mcpRequestHandler.getSseEndpoint();
-                if (sseEndpoint.equals(path) || messageEndpoint.equals(path)) {
+                String mcpEndpoint = mcpRequestHandler.getMcpEndpoint();
+                if (mcpEndpoint.equals(path)) {
                     mcpRequestHandler.handle(ctx, request);
                     isMcpHandled = true;
                     return;

core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/HttpRequestHandler.java

@@ -41,8 +41,11 @@
             <artifactId>jackson-databind</artifactId>
             <version>2.18.1</version>
         </dependency>
-
-
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+            <version>2.18.1</version>
+        </dependency>
 
         <!-- log dependencies -->
         <dependency>

labs/arthas-mcp-server/pom.xml

@@ -29,11 +29,18 @@ public CommandExecutor getCommandExecutor() {
     }
 
     public ArthasMcpServer start() {
+        logger.info("Initializing Arthas MCP Bootstrap...");
         try {
-            // Create and start MCP server
-            mcpServer = new ArthasMcpServer();
+            logger.debug("Creating MCP server instance with command executor: {}", 
+                    commandExecutor.getClass().getSimpleName());
+            
+            // Create and start MCP server with CommandExecutor
+            mcpServer = new ArthasMcpServer(commandExecutor);
+            logger.debug("MCP server instance created successfully");
+            
             mcpServer.start();
             logger.info("Arthas MCP server initialized successfully");
+            logger.info("Bootstrap ready - server is operational");
             return mcpServer;
         } catch (Exception e) {
             logger.error("Failed to initialize Arthas MCP server", e);
@@ -42,8 +49,14 @@ public ArthasMcpServer start() {
     }
 
     public void shutdown() {
+        logger.info("Initiating Arthas MCP Bootstrap shutdown...");
         if (mcpServer != null) {
+            logger.debug("Stopping MCP server...");
             mcpServer.stop();
+            logger.info("MCP server stopped");
+        } else {
+            logger.warn("MCP server was null during shutdown - may not have been properly initialized");
         }
+        logger.info("Arthas MCP Bootstrap shutdown completed");
     }
 }

labs/arthas-mcp-server/src/main/java/com/taobao/arthas/mcp/server/ArthasMcpBootstrap.java

@@ -4,20 +4,24 @@
 import com.taobao.arthas.mcp.server.protocol.config.McpServerProperties;
 import com.taobao.arthas.mcp.server.protocol.server.McpNettyServer;
 import com.taobao.arthas.mcp.server.protocol.server.McpServer;
-import com.taobao.arthas.mcp.server.protocol.server.handler.McpRequestHandler;
-import com.taobao.arthas.mcp.server.protocol.server.transport.HttpNettyServerTransportProvider;
-import com.taobao.arthas.mcp.server.protocol.spec.McpSchema.*;
-import com.taobao.arthas.mcp.server.protocol.spec.McpServerTransportProvider;
+import com.taobao.arthas.mcp.server.protocol.server.McpStatelessNettyServer;
+import com.taobao.arthas.mcp.server.protocol.server.handler.McpHttpRequestHandler;
+import com.taobao.arthas.mcp.server.protocol.server.handler.McpStatelessHttpRequestHandler;
+import com.taobao.arthas.mcp.server.protocol.server.handler.McpStreamableHttpRequestHandler;
+import com.taobao.arthas.mcp.server.protocol.server.transport.NettyStatelessServerTransport;
+import com.taobao.arthas.mcp.server.protocol.server.transport.NettyStreamableServerTransportProvider;
+import com.taobao.arthas.mcp.server.protocol.spec.McpSchema.Implementation;
+import com.taobao.arthas.mcp.server.protocol.spec.McpSchema.ServerCapabilities;
+import com.taobao.arthas.mcp.server.protocol.spec.McpStreamableServerTransportProvider;
 import com.taobao.arthas.mcp.server.tool.DefaultToolCallbackProvider;
 import com.taobao.arthas.mcp.server.tool.ToolCallback;
 import com.taobao.arthas.mcp.server.tool.ToolCallbackProvider;
 import com.taobao.arthas.mcp.server.tool.util.McpToolUtils;
+import com.taobao.arthas.mcp.server.util.JsonParser;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Arrays;
-import java.util.List;
-import java.util.Objects;
+import java.util.*;
 import java.util.stream.Collectors;
 
 /**
@@ -26,66 +30,106 @@
  */
 public class ArthasMcpServer {
     private static final Logger logger = LoggerFactory.getLogger(ArthasMcpServer.class);
-    private McpNettyServer server;
+
+    private McpNettyServer streamableServer;
+    private McpStatelessNettyServer statelessServer;
+
     private final int port;
     private final String bindAddress;
-    private McpRequestHandler mcpRequestHandler;
+
+    private final CommandExecutor commandExecutor;
+
+    private McpHttpRequestHandler unifiedMcpHandler;
+
+    private McpStreamableHttpRequestHandler streamableHandler;
+
+    private McpStatelessHttpRequestHandler statelessHandler;
 
-    public ArthasMcpServer() {
-        this(8080, "localhost");
+    public ArthasMcpServer(CommandExecutor commandExecutor) {
+        this(8080, "localhost", commandExecutor);
     }
 
-    public ArthasMcpServer(int port, String bindAddress) {
+    public ArthasMcpServer(int port, String bindAddress, CommandExecutor commandExecutor) {
         this.port = port;
         this.bindAddress = bindAddress;
+        this.commandExecutor = commandExecutor;
     }
 
-    public McpRequestHandler getMcpRequestHandler() {
-        return mcpRequestHandler;
+    public McpHttpRequestHandler getMcpRequestHandler() {
+        return unifiedMcpHandler;
     }
 
     /**
      * Start MCP server
      */
     public void start() {
+        logger.info("Starting Arthas MCP server on {}:{}", bindAddress, port);
         try {
-            // 1. Create server configuration
             McpServerProperties properties = new McpServerProperties.Builder()
                     .name("arthas-mcp-server")
                     .version("1.0.0")
                     .bindAddress(bindAddress)
                     .port(port)
-                    .messageEndpoint("/sse/message")
-                    .sseEndpoint("/sse")
+                    .mcpEndpoint("/mcp")
                     .toolChangeNotification(true)
                     .resourceChangeNotification(true)
                     .promptChangeNotification(true)
-                    .objectMapper(new ObjectMapper())
+                    .objectMapper(JsonParser.getObjectMapper())
                     .build();
-            
-            // 2. Create transport provider
-            McpServerTransportProvider transportProvider = createHttpTransportProvider(properties);
-            mcpRequestHandler = transportProvider.getMcpRequestHandler();
-            
-            // 3. Create server builder
-            McpServer.NettySpecification serverBuilder = McpServer.netty(transportProvider)
-                    .serverInfo(new Implementation(properties.getName(), properties.getVersion()))
-                    .capabilities(buildServerCapabilities(properties))
-                    .instructions(properties.getInstructions())
-                    .requestTimeout(properties.getRequestTimeout())
-                    .objectMapper(properties.getObjectMapper() != null ? properties.getObjectMapper() : new ObjectMapper());
 
             ToolCallbackProvider toolCallbackProvider = new DefaultToolCallbackProvider();
             ToolCallback[] callbacks = toolCallbackProvider.getToolCallbacks();
             List<ToolCallback> providerToolCallbacks = Arrays.stream(callbacks)
                     .filter(Objects::nonNull)
                     .collect(Collectors.toList());
 
-            serverBuilder.tools(
-                    McpToolUtils.toToolSpecifications(providerToolCallbacks, properties));
-            server = serverBuilder.build();
+            // Create transport for both streamable and stateless servers
+            McpStreamableServerTransportProvider transportProvider = createStreamableHttpTransportProvider(properties);
+            streamableHandler = transportProvider.getMcpRequestHandler();
+
+            NettyStatelessServerTransport statelessTransport = createStatelessHttpTransport(properties);
+            statelessHandler = statelessTransport.getMcpRequestHandler();
+
+            unifiedMcpHandler = McpHttpRequestHandler.builder()
+                    .mcpEndpoint(properties.getMcpEndpoint())
+                    .objectMapper(properties.getObjectMapper())
+                    .tools(Arrays.asList(callbacks))
+                    .build();
+            unifiedMcpHandler.setStreamableHandler(streamableHandler);
+            unifiedMcpHandler.setStatelessHandler(statelessHandler);
+
+            // Set up unified MCP handler for both streamable and stateless servers
+            McpServer.StreamableServerNettySpecification streamableServerNettySpecification = McpServer.netty(transportProvider)
+                    .serverInfo(new Implementation(properties.getName(), properties.getVersion()))
+                    .capabilities(buildServerCapabilities(properties))
+                    .instructions(properties.getInstructions())
+                    .requestTimeout(properties.getRequestTimeout())
+                    .commandExecutor(commandExecutor)
+                    .objectMapper(properties.getObjectMapper() != null ? properties.getObjectMapper() : JsonParser.getObjectMapper());
+
+            // Set up unified MCP handler for both streamable and stateless servers
+            McpServer.StatelessServerNettySpecification statelessServerNettySpecification = McpServer.netty(statelessTransport)
+                    .serverInfo(new Implementation(properties.getName(), properties.getVersion()))
+                    .capabilities(buildServerCapabilities(properties))
+                    .instructions(properties.getInstructions())
+                    .requestTimeout(properties.getRequestTimeout())
+                    .commandExecutor(commandExecutor)
+                    .objectMapper(properties.getObjectMapper() != null ? properties.getObjectMapper() : JsonParser.getObjectMapper());
+
+            streamableServerNettySpecification.tools(
+                    McpToolUtils.toStreamableToolSpecifications(providerToolCallbacks));
+            statelessServerNettySpecification.tools(
+                    McpToolUtils.toStatelessToolSpecifications(providerToolCallbacks));
+
+            streamableServer = streamableServerNettySpecification.build();
+            statelessServer = statelessServerNettySpecification.build();
 
-            logger.info("Arthas MCP server started, listening on {}:{}", bindAddress, port);
+            logger.info("Arthas MCP server started successfully");
+            logger.info("- Listening on {}:{}", bindAddress, port);
+            logger.info("- MCP Endpoint: {}", properties.getMcpEndpoint());
+            logger.info("- Transport modes: Streamable + Stateless");
+            logger.info("- Available tools: {}", providerToolCallbacks.size());
+            logger.info("- Server ready to accept connections");
         } catch (Exception e) {
             logger.error("Failed to start Arthas MCP server", e);
             throw new RuntimeException("Failed to start Arthas MCP server", e);
@@ -95,47 +139,52 @@ public void start() {
     /**
      * Create HTTP transport provider
      */
-    private HttpNettyServerTransportProvider createHttpTransportProvider(McpServerProperties properties) {
-        return HttpNettyServerTransportProvider.builder()
-                .messageEndpoint(properties.getMessageEndpoint())
-                .sseEndpoint(properties.getSseEndpoint())
+    private NettyStreamableServerTransportProvider createStreamableHttpTransportProvider(McpServerProperties properties) {
+        return NettyStreamableServerTransportProvider.builder()
+                .mcpEndpoint(properties.getMcpEndpoint())
+                .objectMapper(properties.getObjectMapper() != null ? properties.getObjectMapper() : new ObjectMapper())
+                .build();
+    }
+
+    private NettyStatelessServerTransport createStatelessHttpTransport(McpServerProperties properties) {
+        return NettyStatelessServerTransport.builder()
+                .mcpEndpoint(properties.getMcpEndpoint())
                 .objectMapper(properties.getObjectMapper() != null ? properties.getObjectMapper() : new ObjectMapper())
                 .build();
     }
 
-    /**
-     * Build server capabilities configuration
-     */
     private ServerCapabilities buildServerCapabilities(McpServerProperties properties) {
         return ServerCapabilities.builder()
                 .prompts(new ServerCapabilities.PromptCapabilities(properties.isPromptChangeNotification()))
                 .resources(new ServerCapabilities.ResourceCapabilities(properties.isResourceSubscribe(), properties.isResourceChangeNotification()))
                 .tools(new ServerCapabilities.ToolCapabilities(properties.isToolChangeNotification()))
                 .build();
     }
-    
-    /**
-     * Stop MCP server
-     */
+
     public void stop() {
-        if (server != null) {
-            try {
-                server.closeGracefully().get();
-                logger.info("Arthas MCP server stopped");
-            } catch (Exception e) {
-                logger.error("Failed to stop Arthas MCP server", e);
+        logger.info("Stopping Arthas MCP server...");
+        try {
+            if (unifiedMcpHandler != null) {
+                logger.debug("Shutting down unified MCP handler");
+                unifiedMcpHandler.closeGracefully().get();
+                logger.info("Unified MCP handler stopped successfully");
             }
-        }
-    }
 
-    public static void main(String[] args) {
-        ArthasMcpServer arthasMcpServer = new ArthasMcpServer();
-        arthasMcpServer.start();
-        // Keep the server running
-        try {
-            Thread.sleep(Long.MAX_VALUE);
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
+            if (streamableServer != null) {
+                logger.debug("Shutting down streamable server");
+                streamableServer.closeGracefully().get();
+                logger.info("Streamable server stopped successfully");
+            }
+
+            if (statelessServer != null) {
+                logger.debug("Shutting down stateless server");
+                statelessServer.closeGracefully().get();
+                logger.info("Stateless server stopped successfully");
+            }
+            
+            logger.info("Arthas MCP server stopped completely");
+        } catch (Exception e) {
+            logger.error("Failed to stop Arthas MCP server gracefully", e);
         }
     }
 }

labs/arthas-mcp-server/src/main/java/com/taobao/arthas/mcp/server/ArthasMcpServer.java

@@ -2,8 +2,27 @@
 
 import java.util.Map;
 
-
+/**
+ * 命令执行器接口
+ *
+ * @author Yeaury 2025/5/26
+ */
 public interface CommandExecutor {
 
-    Map<String, Object> execute(String commandLine, long timeout);
+    default Map<String, Object> executeSync(String commandLine, long timeout) {
+        return executeSync(commandLine, timeout, null);
+    }
+
+    Map<String, Object> executeSync(String commandLine, long timeout, String sessionId);
+
+    Map<String, Object> executeAsync(String commandLine, String sessionId);
+
+    Map<String, Object> pullResults(String sessionId, String consumerId);
+
+    Map<String, Object> interruptJob(String sessionId);
+
+    Map<String, Object> createSession();
+
+    Map<String, Object> closeSession(String sessionId);
 }
+