Package com.tnsai.mcp.client

Class StreamableHttpTransport

java.lang.Object

com.tnsai.mcp.client.StreamableHttpTransport

All Implemented Interfaces:

McpTransport

public class StreamableHttpTransport
extends Object
implements McpTransport

Streamable HTTP transport implementation for MCP (spec 2025-03-26).

Unlike the legacy SSE transport which uses separate SSE and message endpoints, Streamable HTTP uses a single endpoint for all communication via POST. Responses can be either JSON (application/json) or SSE (text/event-stream).

 ┌─────────────────────────────────────────────────────────────────────────┐
 │               Streamable HTTP Transport Architecture                    │
 └─────────────────────────────────────────────────────────────────────────┘

   TnsAI Agent      StreamableHttpTransport         MCP Server
       │                     │                          │
       │  connect()          │                          │
       │ ───────────────────►│                          │
       │                     │  POST /mcp (initialize)  │
       │                     │ ────────────────────────►│
       │                     │                          │
       │                     │  Mcp-Session-Id: abc123  │
       │                     │  JSON or SSE response    │
       │                     │ ◄────────────────────────│
       │                     │                          │
       │  send(message)      │                          │
       │ ───────────────────►│  POST /mcp               │
       │                     │  Mcp-Session-Id: abc123  │
       │                     │ ────────────────────────►│
       │                     │                          │
       │                     │  JSON or SSE response    │
       │  onMessage(data)    │ ◄────────────────────────│
       │ ◄───────────────────│                          │
       │                     │                          │
       │                     │  GET /mcp (SSE stream)   │
       │                     │  Mcp-Session-Id: abc123  │
       │                     │ ────────────────────────►│
       │                     │                          │
       │                     │  SSE notifications       │
       │  onMessage(data)    │ ◄────────────────────────│
       │ ◄───────────────────│                          │
       │                     │                          │
       │  disconnect()       │  DELETE /mcp             │
       │ ───────────────────►│  Mcp-Session-Id: abc123  │
       │                     │ ────────────────────────►│
       │                     │                          │
 

Usage

 StreamableHttpTransport transport = StreamableHttpTransport.builder()
     .endpoint("https://api.example.com/mcp")
     .apiKey("your-api-key")
     .build();

 McpClient client = new McpClient(transport);
 client.connect();
 

See Also:

• McpTransport
• McpClient

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	StreamableHttpTransport.Builder	Builder for StreamableHttpTransport.
static class	StreamableHttpTransport.SessionExpiredException	Thrown when the server returns 404 indicating the session has expired.

Method Summary

Modifier and Type	Method	Description
static StreamableHttpTransport.Builder	builder()	Creates a new StreamableHttpTransport builder.
void	connect()	Connects to the MCP server.
void	disconnect()	Disconnects from the MCP server.
String	getEndpoint()	Returns the configured endpoint.
String	getLastEventId()	Returns the last event ID received.
String	getSessionId()	Returns the current session ID.
boolean	isConnected()	Checks if the transport is currently connected.
boolean	isNotificationStreamActive()	Returns whether the notification stream is currently active.
void	openNotificationStream()	Opens a long-lived GET request to receive server-initiated notifications via SSE.
void	send(String message)	Sends a message to the MCP server.
void	setOnError(Consumer<Throwable> errorConsumer)	Sets a consumer for transport errors.
void	setOnMessage(Consumer<String> messageConsumer)	Sets a consumer for incoming messages from the MCP server.
static StreamableHttpTransport	withApiKey(String endpoint, String apiKey)	Creates a StreamableHttpTransport with API key authentication.

Methods inherited from class java.lang.Object

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

Method Details

builder

public static StreamableHttpTransport.Builder builder()

Creates a new StreamableHttpTransport builder.

Returns:

a new Builder instance

withApiKey

public static StreamableHttpTransport withApiKey(String endpoint,
 String apiKey)

Creates a StreamableHttpTransport with API key authentication.

Parameters:

endpoint - the MCP server endpoint URL

apiKey - the API key for authentication

Returns:

a new StreamableHttpTransport instance

connect

public void connect()
             throws IOException

Description copied from interface: McpTransport

Connects to the MCP server.

Specified by:

connect in interface McpTransport

Throws:

IOException - if connection fails

send

public void send(String message)
          throws IOException

Description copied from interface: McpTransport

Sends a message to the MCP server.

Specified by:

send in interface McpTransport

Parameters:

message - the message to send (JSON-RPC request)

Throws:

IOException - if sending fails

openNotificationStream

public void openNotificationStream()
                            throws IOException

Opens a long-lived GET request to receive server-initiated notifications via SSE.

This runs in a background thread and delivers notifications through the setOnMessage(Consumer) callback. Requires an established session (i.e., a successful initialize must have been sent first).

Throws:

IOException - if the transport is not connected or no session exists

setOnMessage

public void setOnMessage(Consumer<String> messageConsumer)

Description copied from interface: McpTransport

Sets a consumer for incoming messages from the MCP server.

Specified by:

setOnMessage in interface McpTransport

Parameters:

messageConsumer - the consumer for incoming messages (JSON-RPC response/notification)

setOnError

public void setOnError(Consumer<Throwable> errorConsumer)

Description copied from interface: McpTransport

Sets a consumer for transport errors.

Specified by:

setOnError in interface McpTransport

Parameters:

errorConsumer - the consumer for transport errors

disconnect

public void disconnect()
                throws IOException

Description copied from interface: McpTransport

Disconnects from the MCP server.

Specified by:

disconnect in interface McpTransport

Throws:

IOException - if disconnection fails

isConnected

public boolean isConnected()

Description copied from interface: McpTransport

Checks if the transport is currently connected.

Specified by:

isConnected in interface McpTransport

Returns:

true if connected, false otherwise

getEndpoint

public String getEndpoint()

Returns the configured endpoint.

Returns:

the endpoint URL

getSessionId

public String getSessionId()

Returns the current session ID.

Returns:

the session ID, or null if no session established

getLastEventId

public String getLastEventId()

Returns the last event ID received.

Returns:

the last event ID, or null if none received

isNotificationStreamActive

public boolean isNotificationStreamActive()

Returns whether the notification stream is currently active.

Returns:

true if the GET notification stream is running