Package io.modelcontextprotocol.client.transport

Class WebFluxSseClientTransport

java.lang.Object
io.modelcontextprotocol.client.transport.WebFluxSseClientTransport
All Implemented Interfaces:
io.modelcontextprotocol.spec.McpClientTransport, io.modelcontextprotocol.spec.McpTransport
public class WebFluxSseClientTransport extends Object implements io.modelcontextprotocol.spec.McpClientTransport
Server-Sent Events (SSE) implementation of the McpTransport that follows the MCP HTTP with SSE transport specification.

This transport establishes a bidirectional communication channel where:

  • Inbound messages are received through an SSE connection from the server
  • Outbound messages are sent via HTTP POST requests to a server-provided endpoint

The message flow follows these steps:

  1. The client establishes an SSE connection to the server's /sse endpoint
  2. The server sends an 'endpoint' event containing the URI for sending messages
This implementation uses WebClient for HTTP communications and supports JSON serialization/deserialization of messages.
Author:
Christian Tzolov
See Also:

  • Nested Class Summary

    Nested Classes
    Modifier and Type
    Class
    Description
    static class 
    WebFluxSseClientTransport.Builder
    Builder for WebFluxSseClientTransport.

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    protected io.modelcontextprotocol.json.McpJsonMapper
    jsonMapper
    JSON mapper for serializing outbound messages and deserializing inbound messages.
    protected final reactor.core.publisher.Sinks.One<String>
    messageEndpointSink
    Sink for managing the message endpoint URI provided by the server.

  • Constructor Summary

    Constructors
    Constructor
    Description
    WebFluxSseClientTransport(org.springframework.web.reactive.function.client.WebClient.Builder webClientBuilder, io.modelcontextprotocol.json.McpJsonMapper jsonMapper)
    Constructs a new SseClientTransport with the specified WebClient builder and ObjectMapper.
    WebFluxSseClientTransport(org.springframework.web.reactive.function.client.WebClient.Builder webClientBuilder, io.modelcontextprotocol.json.McpJsonMapper jsonMapper, String sseEndpoint)
    Constructs a new SseClientTransport with the specified WebClient builder and ObjectMapper.

  • Method Summary

    Modifier and Type
    Method
    Description
    static WebFluxSseClientTransport.Builder
    builder(org.springframework.web.reactive.function.client.WebClient.Builder webClientBuilder)
    Creates a new builder for WebFluxSseClientTransport.
    reactor.core.publisher.Mono<Void>
    closeGracefully()
    Implements graceful shutdown of the transport.
    reactor.core.publisher.Mono<Void>
    connect(Function<reactor.core.publisher.Mono<io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage>,reactor.core.publisher.Mono<io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage>> handler)
    Establishes a connection to the MCP server using Server-Sent Events (SSE).
    protected reactor.core.publisher.Flux<org.springframework.http.codec.ServerSentEvent<String>>
    eventStream()
    Initializes and starts the inbound SSE event processing.
    List<String>
    protocolVersions()
     
    reactor.core.publisher.Mono<Void>
    sendMessage(io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage message)
    Sends a JSON-RPC message to the server using the endpoint provided during connection.
    <T> T
    unmarshalFrom(Object data, io.modelcontextprotocol.json.TypeRef<T> typeRef)
    Unmarshalls data from a generic Object into the specified type using the configured ObjectMapper.

    Methods inherited from class java.lang.Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

    Methods inherited from interface io.modelcontextprotocol.spec.McpClientTransport

    setExceptionHandler

    Methods inherited from interface io.modelcontextprotocol.spec.McpTransport

    close

  • Field Details

    • jsonMapper

      protected io.modelcontextprotocol.json.McpJsonMapper jsonMapper
      JSON mapper for serializing outbound messages and deserializing inbound messages. Handles conversion between JSON-RPC messages and their string representation.

    • messageEndpointSink

      protected final reactor.core.publisher.Sinks.One<String> messageEndpointSink
      Sink for managing the message endpoint URI provided by the server. Stores the most recent endpoint URI and makes it available for outbound message processing.

  • Constructor Details

    • WebFluxSseClientTransport

      public WebFluxSseClientTransport(org.springframework.web.reactive.function.client.WebClient.Builder webClientBuilder, io.modelcontextprotocol.json.McpJsonMapper jsonMapper)
      Constructs a new SseClientTransport with the specified WebClient builder and ObjectMapper. Initializes both inbound and outbound message processing pipelines.
      Parameters:
      webClientBuilder - the WebClient.Builder to use for creating the WebClient instance
      jsonMapper - the ObjectMapper to use for JSON processing
      Throws:
      IllegalArgumentException - if either parameter is null

    • WebFluxSseClientTransport

      public WebFluxSseClientTransport(org.springframework.web.reactive.function.client.WebClient.Builder webClientBuilder, io.modelcontextprotocol.json.McpJsonMapper jsonMapper, String sseEndpoint)
      Constructs a new SseClientTransport with the specified WebClient builder and ObjectMapper. Initializes both inbound and outbound message processing pipelines.
      Parameters:
      webClientBuilder - the WebClient.Builder to use for creating the WebClient instance
      jsonMapper - the ObjectMapper to use for JSON processing
      sseEndpoint - the SSE endpoint URI to use for establishing the connection
      Throws:
      IllegalArgumentException - if either parameter is null

  • Method Details

    • protocolVersions

      public List<String> protocolVersions()
      Specified by:
      protocolVersions in interface io.modelcontextprotocol.spec.McpTransport

    • connect

      public reactor.core.publisher.Mono<Void> connect(Function<reactor.core.publisher.Mono<io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage>,reactor.core.publisher.Mono<io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage>> handler)
      Establishes a connection to the MCP server using Server-Sent Events (SSE). This method initiates the SSE connection and sets up the message processing pipeline.

      The connection process follows these steps:

      1. Establishes an SSE connection to the server's /sse endpoint
      2. Waits for the server to send an 'endpoint' event with the message posting URI
      3. Sets up message handling for incoming JSON-RPC messages

      The connection is considered established only after receiving the endpoint event from the server.

      Specified by:
      connect in interface io.modelcontextprotocol.spec.McpClientTransport
      Parameters:
      handler - a function that processes incoming JSON-RPC messages and returns responses
      Returns:
      a Mono that completes when the connection is fully established

    • sendMessage

      public reactor.core.publisher.Mono<Void> sendMessage(io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage message)
      Sends a JSON-RPC message to the server using the endpoint provided during connection.

      Messages are sent via HTTP POST requests to the server-provided endpoint URI. The message is serialized to JSON before transmission. If the transport is in the process of closing, the message send operation is skipped gracefully.

      Specified by:
      sendMessage in interface io.modelcontextprotocol.spec.McpTransport
      Parameters:
      message - the JSON-RPC message to send
      Returns:
      a Mono that completes when the message has been sent successfully
      Throws:
      RuntimeException - if message serialization fails

    • eventStream

      protected reactor.core.publisher.Flux<org.springframework.http.codec.ServerSentEvent<String>> eventStream()
      Initializes and starts the inbound SSE event processing. Establishes the SSE connection and sets up event handling for both message and endpoint events. Includes automatic retry logic for handling transient connection failures.

    • closeGracefully

      public reactor.core.publisher.Mono<Void> closeGracefully()
      Implements graceful shutdown of the transport. Cleans up all resources including subscriptions and schedulers. Ensures orderly shutdown of both inbound and outbound message processing.
      Specified by:
      closeGracefully in interface io.modelcontextprotocol.spec.McpTransport
      Returns:
      a Mono that completes when shutdown is finished

    • unmarshalFrom

      public <T> T unmarshalFrom(Object data, io.modelcontextprotocol.json.TypeRef<T> typeRef)
      Unmarshalls data from a generic Object into the specified type using the configured ObjectMapper.

      This method is particularly useful when working with JSON-RPC parameters or result objects that need to be converted to specific Java types. It leverages Jackson's type conversion capabilities to handle complex object structures.

      Specified by:
      unmarshalFrom in interface io.modelcontextprotocol.spec.McpTransport
      Type Parameters:
      T - the target type to convert the data into
      Parameters:
      data - the source object to convert
      typeRef - the TypeRef describing the target type
      Returns:
      the unmarshalled object of type T
      Throws:
      IllegalArgumentException - if the conversion cannot be performed

    • builder

      public static WebFluxSseClientTransport.Builder builder(org.springframework.web.reactive.function.client.WebClient.Builder webClientBuilder)
      Creates a new builder for WebFluxSseClientTransport.
      Parameters:
      webClientBuilder - the WebClient.Builder to use for creating the WebClient instance
      Returns:
      a new builder instance