第七章：MCP 使用范式

<dependencies>
    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-starter-mcp-server-webflux</artifactId>
    </dependency>
</dependencies>

server:
  port: 19000

spring:
  application:
    name: mcp-webflux-server
  ai:
    mcp:
      server:
        name: webflux-mcp-server
        version: 1.0.0
        type: ASYNC  # Recommended for reactive applications
        instructions: "This reactive server provides time information tools and resources"
        sse-message-endpoint: /mcp/messages
        capabilities:
          tool: true
          resource: true
          prompt: true
          completion: true

package com.spring.ai.tutorial.mcp.server.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.ai.tool.annotation.Tool;
import org.springframework.ai.tool.annotation.ToolParam;
import org.springframework.stereotype.Service;

import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;

@Service
public class TimeService {

    private static final Logger logger = LoggerFactory.getLogger(TimeService.class);

    @Tool(description = "Get the time of a specified city.")
    public String  getCityTimeMethod(@ToolParam(description = "Time zone id, such as Asia/Shanghai") String timeZoneId) {
        logger.info("The current time zone is {}", timeZoneId);
        return String.format("The current time zone is %s and the current time is " + "%s", timeZoneId,
                getTimeByZoneId(timeZoneId));
    }

    private String getTimeByZoneId(String zoneId) {

        // Get the time zone using ZoneId
        ZoneId zid = ZoneId.of(zoneId);

        // Get the current time in this time zone
        ZonedDateTime zonedDateTime = ZonedDateTime.now(zid);

        // Defining a formatter
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss z");

        // Format ZonedDateTime as a string
        String formattedDateTime = zonedDateTime.format(formatter);

        return formattedDateTime;
    }
}

package com.spring.ai.tutorial.mcp.server;

import com.spring.ai.tutorial.mcp.server.service.TimeService;
import org.springframework.ai.tool.ToolCallbackProvider;
import org.springframework.ai.tool.method.MethodToolCallbackProvider;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class WebfluxServerApplication {

    public static void main(String[] args) {
        SpringApplication.run(WebfluxServerApplication.class, args);
    }

    @Bean
    public ToolCallbackProvider timeTools(TimeService timeService) {
        return MethodToolCallbackProvider.builder().toolObjects(timeService).build();
    }
}

<dependencies>

    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-autoconfigure-model-openai</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-autoconfigure-model-chat-client</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.ai</groupId>
        <artifactId>spring-ai-starter-mcp-client-webflux</artifactId>
    </dependency>

</dependencies>

server:
  port: 19100

spring:
  application:
    name: mcp-webflux-client
  main:
    web-application-type: none
  ai:
    openai:
      api-key: ${DASHSCOPEAPIKEY}
      base-url: https://dashscope.aliyuncs.com/compatible-mode
      chat:
        options:
          model: qwen-max
    mcp:
      client:
        enabled: true
        name: my-mcp-client
        version: 1.0.0
        request-timeout: 30s
        type: ASYNC  # or ASYNC for reactive applications
        sse:
          connections:
            server1:
              url: http://localhost:19000

package com.spring.ai.tutorial.mcp.client;

import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.tool.ToolCallbackProvider;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;

import java.util.Scanner;

@SpringBootApplication
public class WebfluxClientApplication {

    public static void main(String[] args) {
        SpringApplication.run(WebfluxClientApplication.class, args);
    }

    @Bean
    public CommandLineRunner predefinedQuestions(ChatClient.Builder chatClientBuilder, ToolCallbackProvider tools,
                                                 ConfigurableApplicationContext context) {

        return args -> {
            var chatClient = chatClientBuilder
                    .defaultToolCallbacks(tools.getToolCallbacks())
                    .build();

            Scanner scanner = new Scanner(System.in);
            while (true) {
                System.out.print("\n>>> QUESTION: ");
                String userInput = scanner.nextLine();
                if (userInput.equalsIgnoreCase("exit")) {
                    break;
                }
                System.out.println("\n>>> ASSISTANT: " + chatClient.prompt(userInput).call().content());
            }
            scanner.close();
            context.close();
        };
    }
}

<dependency>
  <groupId>io.modelcontextprotocol.sdk</groupId>
  <artifactId>mcp</artifactId>
  <version>0.10.0</version>
</dependency>

package io.modelcontextprotocol.spec;

import com.fasterxml.jackson.core.type.TypeReference;
import reactor.core.publisher.Mono;

public interface McpTransport {
    default void close() {
        this.closeGracefully().subscribe();
    }

    Mono<Void> closeGracefully();

    Mono<Void> sendMessage(McpSchema.JSONRPCMessage message);

    <T> T unmarshalFrom(Object data, TypeReference<T> typeRef);
}

package io.modelcontextprotocol.spec;

import java.util.function.Function;
import reactor.core.publisher.Mono;

public interface McpClientTransport extends McpTransport {
    Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler);
}

package io.modelcontextprotocol.client.transport;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.util.Assert;
import io.modelcontextprotocol.util.Utils;
import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpClient.Version;
import java.net.http.HttpRequest.BodyPublishers;
import java.net.http.HttpResponse.BodyHandlers;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;

public class HttpClientSseClientTransport implements McpClientTransport {
    private static final Logger logger = LoggerFactory.getLogger(HttpClientSseClientTransport.class);
    private static final String MESSAGEEVENTTYPE = "message";
    private static final String ENDPOINTEVENTTYPE = "endpoint";
    private static final String DEFAULTSSEENDPOINT = "/sse";
    private final URI baseUri;
    private final String sseEndpoint;
    private final FlowSseClient sseClient;
    private final HttpClient httpClient;
    private final HttpRequest.Builder requestBuilder;
    protected ObjectMapper objectMapper;
    private volatile boolean isClosing;
    private final CountDownLatch closeLatch;
    private final AtomicReference<String> messageEndpoint;
    private final AtomicReference<CompletableFuture<Void>> connectionFuture;

    /** @deprecated */
    @Deprecated(
        forRemoval = true
    )
    public HttpClientSseClientTransport(String baseUri) {
        this(HttpClient.newBuilder(), baseUri, new ObjectMapper());
    }

    /** @deprecated */
    @Deprecated(
        forRemoval = true
    )
    public HttpClientSseClientTransport(HttpClient.Builder clientBuilder, String baseUri, ObjectMapper objectMapper) {
        this(clientBuilder, baseUri, "/sse", objectMapper);
    }

    /** @deprecated */
    @Deprecated(
        forRemoval = true
    )
    public HttpClientSseClientTransport(HttpClient.Builder clientBuilder, String baseUri, String sseEndpoint, ObjectMapper objectMapper) {
        this(clientBuilder, HttpRequest.newBuilder(), baseUri, sseEndpoint, objectMapper);
    }

    /** @deprecated */
    @Deprecated(
        forRemoval = true
    )
    public HttpClientSseClientTransport(HttpClient.Builder clientBuilder, HttpRequest.Builder requestBuilder, String baseUri, String sseEndpoint, ObjectMapper objectMapper) {
        this(clientBuilder.connectTimeout(Duration.ofSeconds(10L)).build(), requestBuilder, baseUri, sseEndpoint, objectMapper);
    }

    HttpClientSseClientTransport(HttpClient httpClient, HttpRequest.Builder requestBuilder, String baseUri, String sseEndpoint, ObjectMapper objectMapper) {
        this.isClosing = false;
        this.closeLatch = new CountDownLatch(1);
        this.messageEndpoint = new AtomicReference();
        this.connectionFuture = new AtomicReference();
        Assert.notNull(objectMapper, "ObjectMapper must not be null");
        Assert.hasText(baseUri, "baseUri must not be empty");
        Assert.hasText(sseEndpoint, "sseEndpoint must not be empty");
        Assert.notNull(httpClient, "httpClient must not be null");
        Assert.notNull(requestBuilder, "requestBuilder must not be null");
        this.baseUri = URI.create(baseUri);
        this.sseEndpoint = sseEndpoint;
        this.objectMapper = objectMapper;
        this.httpClient = httpClient;
        this.requestBuilder = requestBuilder;
        this.sseClient = new FlowSseClient(this.httpClient, requestBuilder);
    }

    public static Builder builder(String baseUri) {
        return (new Builder()).baseUri(baseUri);
    }

    public Mono<Void> connect(final Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
        final CompletableFuture<Void> future = new CompletableFuture();
        this.connectionFuture.set(future);
        URI clientUri = Utils.resolveUri(this.baseUri, this.sseEndpoint);
        this.sseClient.subscribe(clientUri.toString(), new FlowSseClient.SseEventHandler() {
            public void onEvent(FlowSseClient.SseEvent event) {
                if (!HttpClientSseClientTransport.this.isClosing) {
                    try {
                        if ("endpoint".equals(event.type())) {
                            String endpoint = event.data();
                            HttpClientSseClientTransport.this.messageEndpoint.set(endpoint);
                            HttpClientSseClientTransport.this.closeLatch.countDown();
                            future.complete((Object)null);
                        } else if ("message".equals(event.type())) {
                            McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(HttpClientSseClientTransport.this.objectMapper, event.data());
                            ((Mono)handler.apply(Mono.just(message))).subscribe();
                        } else {
                            HttpClientSseClientTransport.logger.error("Received unrecognized SSE event type: {}", event.type());
                        }
                    } catch (IOException e) {
                        HttpClientSseClientTransport.logger.error("Error processing SSE event", e);
                        future.completeExceptionally(e);
                    }

                }
            }

            public void onError(Throwable error) {
                if (!HttpClientSseClientTransport.this.isClosing) {
                    HttpClientSseClientTransport.logger.error("SSE connection error", error);
                    future.completeExceptionally(error);
                }

            }
        });
        return Mono.fromFuture(future);
    }

    public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
        if (this.isClosing) {
            return Mono.empty();
        } else {
            try {
                if (!this.closeLatch.await(10L, TimeUnit.SECONDS)) {
                    return Mono.error(new McpError("Failed to wait for the message endpoint"));
                }
            } catch (InterruptedException var6) {
                return Mono.error(new McpError("Failed to wait for the message endpoint"));
            }

            String endpoint = (String)this.messageEndpoint.get();
            if (endpoint == null) {
                return Mono.error(new McpError("No message endpoint available"));
            } else {
                try {
                    String jsonText = this.objectMapper.writeValueAsString(message);
                    URI requestUri = Utils.resolveUri(this.baseUri, endpoint);
                    HttpRequest request = this.requestBuilder.uri(requestUri).POST(BodyPublishers.ofString(jsonText)).build();
                    return Mono.fromFuture(this.httpClient.sendAsync(request, BodyHandlers.discarding()).thenAccept((response) -> {
                        if (response.statusCode() != 200 && response.statusCode() != 201 && response.statusCode() != 202 && response.statusCode() != 206) {
                            logger.error("Error sending message: {}", response.statusCode());
                        }

                    }));
                } catch (IOException e) {
                    return !this.isClosing ? Mono.error(new RuntimeException("Failed to serialize message", e)) : Mono.empty();
                }
            }
        }
    }

    public Mono<Void> closeGracefully() {
        return Mono.fromRunnable(() -> {
            this.isClosing = true;
            CompletableFuture<Void> future = (CompletableFuture)this.connectionFuture.get();
            if (future != null && !future.isDone()) {
                future.cancel(true);
            }

        });
    }

    public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
        return (T)this.objectMapper.convertValue(data, typeRef);
    }

    public static class Builder {
        private String baseUri;
        private String sseEndpoint = "/sse";
        private HttpClient.Builder clientBuilder;
        private ObjectMapper objectMapper;
        private HttpRequest.Builder requestBuilder;

        Builder() {
            this.clientBuilder = HttpClient.newBuilder().version(Version.HTTP11).connectTimeout(Duration.ofSeconds(10L));
            this.objectMapper = new ObjectMapper();
            this.requestBuilder = HttpRequest.newBuilder().header("Content-Type", "application/json");
        }

        /** @deprecated */
        @Deprecated(
            forRemoval = true
        )
        public Builder(String baseUri) {
            this.clientBuilder = HttpClient.newBuilder().version(Version.HTTP11).connectTimeout(Duration.ofSeconds(10L));
            this.objectMapper = new ObjectMapper();
            this.requestBuilder = HttpRequest.newBuilder().header("Content-Type", "application/json");
            Assert.hasText(baseUri, "baseUri must not be empty");
            this.baseUri = baseUri;
        }

        Builder baseUri(String baseUri) {
            Assert.hasText(baseUri, "baseUri must not be empty");
            this.baseUri = baseUri;
            return this;
        }

        public Builder sseEndpoint(String sseEndpoint) {
            Assert.hasText(sseEndpoint, "sseEndpoint must not be empty");
            this.sseEndpoint = sseEndpoint;
            return this;
        }

        public Builder clientBuilder(HttpClient.Builder clientBuilder) {
            Assert.notNull(clientBuilder, "clientBuilder must not be null");
            this.clientBuilder = clientBuilder;
            return this;
        }

        public Builder customizeClient(final Consumer<HttpClient.Builder> clientCustomizer) {
            Assert.notNull(clientCustomizer, "clientCustomizer must not be null");
            clientCustomizer.accept(this.clientBuilder);
            return this;
        }

        public Builder requestBuilder(HttpRequest.Builder requestBuilder) {
            Assert.notNull(requestBuilder, "requestBuilder must not be null");
            this.requestBuilder = requestBuilder;
            return this;
        }

        public Builder customizeRequest(final Consumer<HttpRequest.Builder> requestCustomizer) {
            Assert.notNull(requestCustomizer, "requestCustomizer must not be null");
            requestCustomizer.accept(this.requestBuilder);
            return this;
        }

        public Builder objectMapper(ObjectMapper objectMapper) {
            Assert.notNull(objectMapper, "objectMapper must not be null");
            this.objectMapper = objectMapper;
            return this;
        }

        public HttpClientSseClientTransport build() {
            return new HttpClientSseClientTransport(this.clientBuilder.build(), this.requestBuilder, this.baseUri, this.sseEndpoint, this.objectMapper);
        }
    }
}

package io.modelcontextprotocol.client.transport;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.util.Assert;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

public class StdioClientTransport implements McpClientTransport {
    private static final Logger logger = LoggerFactory.getLogger(StdioClientTransport.class);
    private final Sinks.Many<McpSchema.JSONRPCMessage> inboundSink;
    private final Sinks.Many<McpSchema.JSONRPCMessage> outboundSink;
    private Process process;
    private ObjectMapper objectMapper;
    private Scheduler inboundScheduler;
    private Scheduler outboundScheduler;
    private Scheduler errorScheduler;
    private final ServerParameters params;
    private final Sinks.Many<String> errorSink;
    private volatile boolean isClosing;
    private Consumer<String> stdErrorHandler;

    public StdioClientTransport(ServerParameters params) {
        this(params, new ObjectMapper());
    }

    public StdioClientTransport(ServerParameters params, ObjectMapper objectMapper) {
        this.isClosing = false;
        this.stdErrorHandler = (error) -> logger.info("STDERR Message received: {}", error);
        Assert.notNull(params, "The params can not be null");
        Assert.notNull(objectMapper, "The ObjectMapper can not be null");
        this.inboundSink = Sinks.many().unicast().onBackpressureBuffer();
        this.outboundSink = Sinks.many().unicast().onBackpressureBuffer();
        this.params = params;
        this.objectMapper = objectMapper;
        this.errorSink = Sinks.many().unicast().onBackpressureBuffer();
        this.inboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "inbound");
        this.outboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "outbound");
        this.errorScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), "error");
    }

    public Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
        return Mono.fromRunnable(() -> {
            this.handleIncomingMessages(handler);
            this.handleIncomingErrors();
            List<String> fullCommand = new ArrayList();
            fullCommand.add(this.params.getCommand());
            fullCommand.addAll(this.params.getArgs());
            ProcessBuilder processBuilder = this.getProcessBuilder();
            processBuilder.command(fullCommand);
            processBuilder.environment().putAll(this.params.getEnv());

            try {
                this.process = processBuilder.start();
            } catch (IOException e) {
                throw new RuntimeException("Failed to start process with command: " + String.valueOf(fullCommand), e);
            }

            if (this.process.getInputStream() != null && this.process.getOutputStream() != null) {
                this.startInboundProcessing();
                this.startOutboundProcessing();
                this.startErrorProcessing();
            } else {
                this.process.destroy();
                throw new RuntimeException("Process input or output stream is null");
            }
        }).subscribeOn(Schedulers.boundedElastic());
    }

    protected ProcessBuilder getProcessBuilder() {
        return new ProcessBuilder(new String[0]);
    }

    public void setStdErrorHandler(Consumer<String> errorHandler) {
        this.stdErrorHandler = errorHandler;
    }

    public void awaitForExit() {
        try {
            this.process.waitFor();
        } catch (InterruptedException e) {
            throw new RuntimeException("Process interrupted", e);
        }
    }

    private void startErrorProcessing() {
        this.errorScheduler.schedule(() -> {
            try {
                BufferedReader processErrorReader = new BufferedReader(new InputStreamReader(this.process.getErrorStream()));

                String line;
                try {
                    while(!this.isClosing && (line = processErrorReader.readLine()) != null) {
                        try {
                            if (!this.errorSink.tryEmitNext(line).isSuccess()) {
                                if (!this.isClosing) {
                                    logger.error("Failed to emit error message");
                                }
                                break;
                            }
                        } catch (Exception e) {
                            if (!this.isClosing) {
                                logger.error("Error processing error message", e);
                            }
                            break;
                        }
                    }
                } catch (Throwable var12) {
                    try {
                        processErrorReader.close();
                    } catch (Throwable var10) {
                        var12.addSuppressed(var10);
                    }

                    throw var12;
                }

                processErrorReader.close();
            } catch (IOException e) {
                if (!this.isClosing) {
                    logger.error("Error reading from error stream", e);
                }
            } finally {
                this.isClosing = true;
                this.errorSink.tryEmitComplete();
            }

        });
    }

    private void handleIncomingMessages(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> inboundMessageHandler) {
        this.inboundSink.asFlux().flatMap((message) -> Mono.just(message).transform(inboundMessageHandler).contextWrite((ctx) -> ctx.put("observation", "myObservation"))).subscribe();
    }

    private void handleIncomingErrors() {
        this.errorSink.asFlux().subscribe((e) -> this.stdErrorHandler.accept(e));
    }

    public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
        return this.outboundSink.tryEmitNext(message).isSuccess() ? Mono.empty() : Mono.error(new RuntimeException("Failed to enqueue message"));
    }

    private void startInboundProcessing() {
        this.inboundScheduler.schedule(() -> {
            try {
                BufferedReader processReader = new BufferedReader(new InputStreamReader(this.process.getInputStream()));

                String line;
                try {
                    while(!this.isClosing && (line = processReader.readLine()) != null) {
                        try {
                            McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, line);
                            if (!this.inboundSink.tryEmitNext(message).isSuccess()) {
                                if (!this.isClosing) {
                                    logger.error("Failed to enqueue inbound message: {}", message);
                                }
                                break;
                            }
                        } catch (Exception e) {
                            if (!this.isClosing) {
                                logger.error("Error processing inbound message for line: " + line, e);
                            }
                            break;
                        }
                    }
                } catch (Throwable var12) {
                    try {
                        processReader.close();
                    } catch (Throwable var10) {
                        var12.addSuppressed(var10);
                    }

                    throw var12;
                }

                processReader.close();
            } catch (IOException e) {
                if (!this.isClosing) {
                    logger.error("Error reading from input stream", e);
                }
            } finally {
                this.isClosing = true;
                this.inboundSink.tryEmitComplete();
            }

        });
    }

    private void startOutboundProcessing() {
        this.handleOutbound((messages) -> messages.publishOn(this.outboundScheduler).handle((message, s) -> {
                if (message != null && !this.isClosing) {
                    try {
                        String jsonMessage = this.objectMapper.writeValueAsString(message);
                        jsonMessage = jsonMessage.replace("\r\n", "\\n").replace("\n", "\\n").replace("\r", "\\n");
                        OutputStream os = this.process.getOutputStream();
                        synchronized(os) {
                            os.write(jsonMessage.getBytes(StandardCharsets.UTF8));
                            os.write("\n".getBytes(StandardCharsets.UTF8));
                            os.flush();
                        }

                        s.next(message);
                    } catch (IOException e) {
                        s.error(new RuntimeException(e));
                    }
                }

            }));
    }

    protected void handleOutbound(Function<Flux<McpSchema.JSONRPCMessage>, Flux<McpSchema.JSONRPCMessage>> outboundConsumer) {
        ((Flux)outboundConsumer.apply(this.outboundSink.asFlux())).doOnComplete(() -> {
            this.isClosing = true;
            this.outboundSink.tryEmitComplete();
        }).doOnError((e) -> {
            if (!this.isClosing) {
                logger.error("Error in outbound processing", e);
                this.isClosing = true;
                this.outboundSink.tryEmitComplete();
            }

        }).subscribe();
    }

    public Mono<Void> closeGracefully() {
        return Mono.fromRunnable(() -> {
            this.isClosing = true;
            logger.debug("Initiating graceful shutdown");
        }).then(Mono.defer(() -> {
            this.inboundSink.tryEmitComplete();
            this.outboundSink.tryEmitComplete();
            this.errorSink.tryEmitComplete();
            return Mono.delay(Duration.ofMillis(100L));
        })).then(Mono.defer(() -> {
            logger.debug("Sending TERM to process");
            if (this.process != null) {
                this.process.destroy();
                return Mono.fromFuture(this.process.onExit());
            } else {
                logger.warn("Process not started");
                return Mono.empty();
            }
        })).doOnNext((process) -> {
            if (process.exitValue() != 0) {
                logger.warn("Process terminated with code " + process.exitValue());
            }

        }).then(Mono.fromRunnable(() -> {
            try {
                this.inboundScheduler.dispose();
                this.errorScheduler.dispose();
                this.outboundScheduler.dispose();
                logger.debug("Graceful shutdown completed");
            } catch (Exception e) {
                logger.error("Error during graceful shutdown", e);
            }

        })).then().subscribeOn(Schedulers.boundedElastic());
    }

    public Sinks.Many<String> getErrorSink() {
        return this.errorSink;
    }

    public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
        return (T)this.objectMapper.convertValue(data, typeRef);
    }
}

package io.modelcontextprotocol.client.transport;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.util.Assert;
import java.io.IOException;
import java.util.function.BiConsumer;
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.publisher.SynchronousSink;
import reactor.core.scheduler.Schedulers;
import reactor.util.retry.Retry;

public class WebFluxSseClientTransport implements McpClientTransport {
    private static final Logger logger = LoggerFactory.getLogger(WebFluxSseClientTransport.class);
    private static final String MESSAGEEVENTTYPE = "message";
    private static final String ENDPOINTEVENTTYPE = "endpoint";
    private static final String DEFAULTSSEENDPOINT = "/sse";
    private static final ParameterizedTypeReference<ServerSentEvent<String>> SSETYPE = new ParameterizedTypeReference<ServerSentEvent<String>>() {
    };
    private final WebClient webClient;
    protected ObjectMapper objectMapper;
    private Disposable inboundSubscription;
    private volatile boolean isClosing;
    protected final Sinks.One<String> messageEndpointSink;
    private String sseEndpoint;
    private BiConsumer<Retry.RetrySignal, SynchronousSink<Object>> inboundRetryHandler;

    public WebFluxSseClientTransport(WebClient.Builder webClientBuilder) {
        this(webClientBuilder, new ObjectMapper());
    }

    public WebFluxSseClientTransport(WebClient.Builder webClientBuilder, ObjectMapper objectMapper) {
        this(webClientBuilder, objectMapper, "/sse");
    }

    public WebFluxSseClientTransport(WebClient.Builder webClientBuilder, ObjectMapper objectMapper, String sseEndpoint) {
        this.isClosing = false;
        this.messageEndpointSink = Sinks.one();
        this.inboundRetryHandler = (retrySpec, sink) -> {
            if (this.isClosing) {
                logger.debug("SSE connection closed during shutdown");
                sink.error(retrySpec.failure());
            } else if (retrySpec.failure() instanceof IOException) {
                logger.debug("Retrying SSE connection after IO error");
                sink.next(retrySpec);
            } else {
                logger.error("Fatal SSE error, not retrying: {}", retrySpec.failure().getMessage());
                sink.error(retrySpec.failure());
            }
        };
        Assert.notNull(objectMapper, "ObjectMapper must not be null");
        Assert.notNull(webClientBuilder, "WebClient.Builder must not be null");
        Assert.hasText(sseEndpoint, "SSE endpoint must not be null or empty");
        this.objectMapper = objectMapper;
        this.webClient = webClientBuilder.build();
        this.sseEndpoint = sseEndpoint;
    }

    public Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler) {
        Flux<ServerSentEvent<String>> events = this.eventStream();
        this.inboundSubscription = events.concatMap((event) -> Mono.just(event).handle((e, s) -> {
                if ("endpoint".equals(event.event())) {
                    String messageEndpointUri = (String)event.data();
                    if (this.messageEndpointSink.tryEmitValue(messageEndpointUri).isSuccess()) {
                        s.complete();
                    } else {
                        s.error(new McpError("Failed to handle SSE endpoint event"));
                    }
                } else if ("message".equals(event.event())) {
                    try {
                        McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, (String)event.data());
                        s.next(message);
                    } catch (IOException ioException) {
                        s.error(ioException);
                    }
                } else {
                    s.error(new McpError("Received unrecognized SSE event type: " + event.event()));
                }

            }).transform(handler)).subscribe();
        return this.messageEndpointSink.asMono().then();
    }

    public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
        return this.messageEndpointSink.asMono().flatMap((messageEndpointUri) -> {
            if (this.isClosing) {
                return Mono.empty();
            } else {
                try {
                    String jsonText = this.objectMapper.writeValueAsString(message);
                    return ((WebClient.RequestBodySpec)this.webClient.post().uri(messageEndpointUri, new Object[0])).contentType(MediaType.APPLICATIONJSON).bodyValue(jsonText).retrieve().toBodilessEntity().doOnSuccess((response) -> logger.debug("Message sent successfully")).doOnError((error) -> {
                        if (!this.isClosing) {
                            logger.error("Error sending message: {}", error.getMessage());
                        }

                    });
                } catch (IOException e) {
                    return !this.isClosing ? Mono.error(new RuntimeException("Failed to serialize message", e)) : Mono.empty();
                }
            }
        }).then();
    }

    protected Flux<ServerSentEvent<String>> eventStream() {
        return this.webClient.get().uri(this.sseEndpoint, new Object[0]).accept(new MediaType[]{MediaType.TEXTEVENTSTREAM}).retrieve().bodyToFlux(SSETYPE).retryWhen(Retry.from((retrySignal) -> retrySignal.handle(this.inboundRetryHandler)));
    }

    public Mono<Void> closeGracefully() {
        return Mono.fromRunnable(() -> {
            this.isClosing = true;
            if (this.inboundSubscription != null) {
                this.inboundSubscription.dispose();
            }

        }).then().subscribeOn(Schedulers.boundedElastic());
    }

    public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
        return (T)this.objectMapper.convertValue(data, typeRef);
    }

    public static Builder builder(WebClient.Builder webClientBuilder) {
        return new Builder(webClientBuilder);
    }

    public static class Builder {
        private final WebClient.Builder webClientBuilder;
        private String sseEndpoint = "/sse";
        private ObjectMapper objectMapper = new ObjectMapper();

        public Builder(WebClient.Builder webClientBuilder) {
            Assert.notNull(webClientBuilder, "WebClient.Builder must not be null");
            this.webClientBuilder = webClientBuilder;
        }

        public Builder sseEndpoint(String sseEndpoint) {
            Assert.hasText(sseEndpoint, "sseEndpoint must not be empty");
            this.sseEndpoint = sseEndpoint;
            return this;
        }

        public Builder objectMapper(ObjectMapper objectMapper) {
            Assert.notNull(objectMapper, "objectMapper must not be null");
            this.objectMapper = objectMapper;
            return this;
        }

        public WebFluxSseClientTransport build() {
            return new WebFluxSseClientTransport(this.webClientBuilder, this.objectMapper, this.sseEndpoint);
        }
    }
}

package io.modelcontextprotocol.spec;

public interface McpServerTransport extends McpTransport {
}

package io.modelcontextprotocol.spec;

import reactor.core.publisher.Mono;

public interface McpServerTransportProvider {
    void setSessionFactory(McpServerSession.Factory sessionFactory);

    Mono<Void> notifyClients(String method, Object params);

    default void close() {
        this.closeGracefully().subscribe();
    }

    Mono<Void> closeGracefully();
}

package io.modelcontextprotocol.server.transport;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpServerSession;
import io.modelcontextprotocol.spec.McpServerTransport;
import io.modelcontextprotocol.spec.McpServerTransportProvider;
import io.modelcontextprotocol.util.Assert;
import jakarta.servlet.AsyncContext;
import jakarta.servlet.ServletException;
import jakarta.servlet.annotation.WebServlet;
import jakarta.servlet.http.HttpServlet;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@WebServlet(
    asyncSupported = true
)
public class HttpServletSseServerTransportProvider extends HttpServlet implements McpServerTransportProvider {
    private static final Logger logger = LoggerFactory.getLogger(HttpServletSseServerTransportProvider.class);
    public static final String UTF8 = "UTF-8";
    public static final String APPLICATIONJSON = "application/json";
    public static final String FAILEDTOSENDERRORRESPONSE = "Failed to send error response: {}";
    public static final String DEFAULTSSEENDPOINT = "/sse";
    public static final String MESSAGEEVENTTYPE = "message";
    public static final String ENDPOINTEVENTTYPE = "endpoint";
    public static final String DEFAULTBASEURL = "";
    private final ObjectMapper objectMapper;
    private final String baseUrl;
    private final String messageEndpoint;
    private final String sseEndpoint;
    private final Map<String, McpServerSession> sessions;
    private final AtomicBoolean isClosing;
    private McpServerSession.Factory sessionFactory;

    public HttpServletSseServerTransportProvider(ObjectMapper objectMapper, String messageEndpoint, String sseEndpoint) {
        this(objectMapper, "", messageEndpoint, sseEndpoint);
    }

    public HttpServletSseServerTransportProvider(ObjectMapper objectMapper, String baseUrl, String messageEndpoint, String sseEndpoint) {
        this.sessions = new ConcurrentHashMap();
        this.isClosing = new AtomicBoolean(false);
        this.objectMapper = objectMapper;
        this.baseUrl = baseUrl;
        this.messageEndpoint = messageEndpoint;
        this.sseEndpoint = sseEndpoint;
    }

    public HttpServletSseServerTransportProvider(ObjectMapper objectMapper, String messageEndpoint) {
        this(objectMapper, messageEndpoint, "/sse");
    }

    public void setSessionFactory(McpServerSession.Factory sessionFactory) {
        this.sessionFactory = sessionFactory;
    }

    public Mono<Void> notifyClients(String method, Object params) {
        if (this.sessions.isEmpty()) {
            logger.debug("No active sessions to broadcast message to");
            return Mono.empty();
        } else {
            logger.debug("Attempting to broadcast message to {} active sessions", this.sessions.size());
            return Flux.fromIterable(this.sessions.values()).flatMap((session) -> session.sendNotification(method, params).doOnError((e) -> logger.error("Failed to send message to session {}: {}", session.getId(), e.getMessage())).onErrorComplete()).then();
        }
    }

    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        String requestURI = request.getRequestURI();
        if (!requestURI.endsWith(this.sseEndpoint)) {
            response.sendError(404);
        } else if (this.isClosing.get()) {
            response.sendError(503, "Server is shutting down");
        } else {
            response.setContentType("text/event-stream");
            response.setCharacterEncoding("UTF-8");
            response.setHeader("Cache-Control", "no-cache");
            response.setHeader("Connection", "keep-alive");
            response.setHeader("Access-Control-Allow-Origin", "*");
            String sessionId = UUID.randomUUID().toString();
            AsyncContext asyncContext = request.startAsync();
            asyncContext.setTimeout(0L);
            PrintWriter writer = response.getWriter();
            HttpServletMcpSessionTransport sessionTransport = new HttpServletMcpSessionTransport(sessionId, asyncContext, writer);
            McpServerSession session = this.sessionFactory.create(sessionTransport);
            this.sessions.put(sessionId, session);
            this.sendEvent(writer, "endpoint", this.baseUrl + this.messageEndpoint + "?sessionId=" + sessionId);
        }
    }

    protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        if (this.isClosing.get()) {
            response.sendError(503, "Server is shutting down");
        } else {
            String requestURI = request.getRequestURI();
            if (!requestURI.endsWith(this.messageEndpoint)) {
                response.sendError(404);
            } else {
                String sessionId = request.getParameter("sessionId");
                if (sessionId == null) {
                    response.setContentType("application/json");
                    response.setCharacterEncoding("UTF-8");
                    response.setStatus(400);
                    String jsonError = this.objectMapper.writeValueAsString(new McpError("Session ID missing in message endpoint"));
                    PrintWriter writer = response.getWriter();
                    writer.write(jsonError);
                    writer.flush();
                } else {
                    McpServerSession session = (McpServerSession)this.sessions.get(sessionId);
                    if (session == null) {
                        response.setContentType("application/json");
                        response.setCharacterEncoding("UTF-8");
                        response.setStatus(404);
                        String jsonError = this.objectMapper.writeValueAsString(new McpError("Session not found: " + sessionId));
                        PrintWriter writer = response.getWriter();
                        writer.write(jsonError);
                        writer.flush();
                    } else {
                        try {
                            BufferedReader reader = request.getReader();
                            StringBuilder body = new StringBuilder();

                            String line;
                            while((line = reader.readLine()) != null) {
                                body.append(line);
                            }

                            McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.objectMapper, body.toString());
                            session.handle(message).block();
                            response.setStatus(200);
                        } catch (Exception var11) {
                            Exception e = var11;
                            logger.error("Error processing message: {}", var11.getMessage());

                            try {
                                McpError mcpError = new McpError(e.getMessage());
                                response.setContentType("application/json");
                                response.setCharacterEncoding("UTF-8");
                                response.setStatus(500);
                                String jsonError = this.objectMapper.writeValueAsString(mcpError);
                                PrintWriter writer = response.getWriter();
                                writer.write(jsonError);
                                writer.flush();
                            } catch (IOException ex) {
                                logger.error("Failed to send error response: {}", ex.getMessage());
                                response.sendError(500, "Error processing message");
                            }
                        }

                    }
                }
            }
        }
    }

    public Mono<Void> closeGracefully() {
        this.isClosing.set(true);
        logger.debug("Initiating graceful shutdown with {} active sessions", this.sessions.size());
        return Flux.fromIterable(this.sessions.values()).flatMap(McpServerSession::closeGracefully).then();
    }

    private void sendEvent(PrintWriter writer, String eventType, String data) throws IOException {
        writer.write("event: " + eventType + "\n");
        writer.write("data: " + data + "\n\n");
        writer.flush();
        if (writer.checkError()) {
            throw new IOException("Client disconnected");
        }
    }

    public void destroy() {
        this.closeGracefully().block();
        super.destroy();
    }

    public static Builder builder() {
        return new Builder();
    }

    private class HttpServletMcpSessionTransport implements McpServerTransport {
        private final String sessionId;
        private final AsyncContext asyncContext;
        private final PrintWriter writer;

        HttpServletMcpSessionTransport(String sessionId, AsyncContext asyncContext, PrintWriter writer) {
            this.sessionId = sessionId;
            this.asyncContext = asyncContext;
            this.writer = writer;
            HttpServletSseServerTransportProvider.logger.debug("Session transport {} initialized with SSE writer", sessionId);
        }

        public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
            return Mono.fromRunnable(() -> {
                try {
                    String jsonText = HttpServletSseServerTransportProvider.this.objectMapper.writeValueAsString(message);
                    HttpServletSseServerTransportProvider.this.sendEvent(this.writer, "message", jsonText);
                    HttpServletSseServerTransportProvider.logger.debug("Message sent to session {}", this.sessionId);
                } catch (Exception e) {
                    HttpServletSseServerTransportProvider.logger.error("Failed to send message to session {}: {}", this.sessionId, e.getMessage());
                    HttpServletSseServerTransportProvider.this.sessions.remove(this.sessionId);
                    this.asyncContext.complete();
                }

            });
        }

        public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
            return (T)HttpServletSseServerTransportProvider.this.objectMapper.convertValue(data, typeRef);
        }

        public Mono<Void> closeGracefully() {
            return Mono.fromRunnable(() -> {
                HttpServletSseServerTransportProvider.logger.debug("Closing session transport: {}", this.sessionId);

                try {
                    HttpServletSseServerTransportProvider.this.sessions.remove(this.sessionId);
                    this.asyncContext.complete();
                    HttpServletSseServerTransportProvider.logger.debug("Successfully completed async context for session {}", this.sessionId);
                } catch (Exception e) {
                    HttpServletSseServerTransportProvider.logger.warn("Failed to complete async context for session {}: {}", this.sessionId, e.getMessage());
                }

            });
        }

        public void close() {
            try {
                HttpServletSseServerTransportProvider.this.sessions.remove(this.sessionId);
                this.asyncContext.complete();
                HttpServletSseServerTransportProvider.logger.debug("Successfully completed async context for session {}", this.sessionId);
            } catch (Exception e) {
                HttpServletSseServerTransportProvider.logger.warn("Failed to complete async context for session {}: {}", this.sessionId, e.getMessage());
            }

        }
    }

    public static class Builder {
        private ObjectMapper objectMapper = new ObjectMapper();
        private String baseUrl = "";
        private String messageEndpoint;
        private String sseEndpoint = "/sse";

        public Builder objectMapper(ObjectMapper objectMapper) {
            Assert.notNull(objectMapper, "ObjectMapper must not be null");
            this.objectMapper = objectMapper;
            return this;
        }

        public Builder baseUrl(String baseUrl) {
            Assert.notNull(baseUrl, "Base URL must not be null");
            this.baseUrl = baseUrl;
            return this;
        }

        public Builder messageEndpoint(String messageEndpoint) {
            Assert.hasText(messageEndpoint, "Message endpoint must not be empty");
            this.messageEndpoint = messageEndpoint;
            return this;
        }

        public Builder sseEndpoint(String sseEndpoint) {
            Assert.hasText(sseEndpoint, "SSE endpoint must not be empty");
            this.sseEndpoint = sseEndpoint;
            return this;
        }

        public HttpServletSseServerTransportProvider build() {
            if (this.objectMapper == null) {
                throw new IllegalStateException("ObjectMapper must be set");
            } else if (this.messageEndpoint == null) {
                throw new IllegalStateException("MessageEndpoint must be set");
            } else {
                return new HttpServletSseServerTransportProvider(this.objectMapper, this.baseUrl, this.messageEndpoint, this.sseEndpoint);
            }
        }
    }
}