modelcontextprotocol/csharp-sdk
Watch
1
0
Fork
You've already forked csharp-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Add client-side Streamable HTTP transport support (#356)

Browse source
This commit is contained in:
Stephen Halter 2025-04-30 17:21:43 -07:00 • committed by GitHub
commit e913d80f05
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 908 additions and 464 deletions
Download patch file Download diff file Expand all files Collapse all files

55
src/ModelContextProtocol/Protocol/Transport/SseClientSessionTransport.cs
View file

@ -87,40 +87,17 @@ internal sealed partial class SseClientSessionTransport : TransportBase
messageId = messageWithId.Id.ToString();
}
var httpRequestMessage = new HttpRequestMessage(HttpMethod.Post, _messageEndpoint)
using var httpRequestMessage = new HttpRequestMessage(HttpMethod.Post, _messageEndpoint)
{
Content = content,
};
CopyAdditionalHeaders(httpRequestMessage.Headers);
StreamableHttpClientSessionTransport.CopyAdditionalHeaders(httpRequestMessage.Headers, _options.AdditionalHeaders);
var response = await _httpClient.SendAsync(httpRequestMessage, cancellationToken).ConfigureAwait(false);
response.EnsureSuccessStatusCode();
var responseContent = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
// Check if the message was an initialize request
if (message is JsonRpcRequest request && request.Method == RequestMethods.Initialize)
{
// If the response is not a JSON-RPC response, it is an SSE message
if (string.IsNullOrEmpty(responseContent) || responseContent.Equals("accepted", StringComparison.OrdinalIgnoreCase))
{
LogAcceptedPost(Name, messageId);
// The response will arrive as an SSE message
}
else
{
JsonRpcResponse initializeResponse = JsonSerializer.Deserialize(responseContent, McpJsonUtilities.JsonContext.Default.JsonRpcResponse) ??
throw new InvalidOperationException("Failed to initialize client");
LogTransportReceivedMessage(Name, messageId);
await WriteMessageAsync(initializeResponse, cancellationToken).ConfigureAwait(false);
LogTransportMessageWritten(Name, messageId);
}
return;
}
// Otherwise, check if the response was accepted (the response will come as an SSE message)
if (string.IsNullOrEmpty(responseContent) || responseContent.Equals("accepted", StringComparison.OrdinalIgnoreCase))
{
LogAcceptedPost(Name, messageId);
@ -177,17 +154,13 @@ internal sealed partial class SseClientSessionTransport : TransportBase
}
}
internal Uri? MessageEndpoint => _messageEndpoint;
internal SseClientTransportOptions Options => _options;
private async Task ReceiveMessagesAsync(CancellationToken cancellationToken)
{
try
{
using var request = new HttpRequestMessage(HttpMethod.Get, _sseEndpoint);
request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
CopyAdditionalHeaders(request.Headers);
StreamableHttpClientSessionTransport.CopyAdditionalHeaders(request.Headers, _options.AdditionalHeaders);
using var response = await _httpClient.SendAsync(
request,
@ -251,15 +224,7 @@ internal sealed partial class SseClientSessionTransport : TransportBase
return;
}
string messageId = "(no id)";
if (message is JsonRpcMessageWithId messageWithId)
{
messageId = messageWithId.Id.ToString();
}
LogTransportReceivedMessage(Name, messageId);
await WriteMessageAsync(message, cancellationToken).ConfigureAwait(false);
LogTransportMessageWritten(Name, messageId);
}
catch (JsonException ex)
{
@ -290,20 +255,6 @@ internal sealed partial class SseClientSessionTransport : TransportBase
_connectionEstablished.TrySetResult(true);
}
private void CopyAdditionalHeaders(HttpRequestHeaders headers)
{
if (_options.AdditionalHeaders is not null)
{
foreach (var header in _options.AdditionalHeaders)
{
if (!headers.TryAddWithoutValidation(header.Key, header.Value))
{
throw new InvalidOperationException($"Failed to add header '{header.Key}' with value '{header.Value}' from {nameof(SseClientTransportOptions.AdditionalHeaders)}.");
}
}
}
}
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} accepted SSE transport POST for message ID '{MessageId}'.")]
private partial void LogAcceptedPost(string endpointName, string messageId);

5
src/ModelContextProtocol/Protocol/Transport/SseClientTransport.cs
View file

@ -57,6 +57,11 @@ public sealed class SseClientTransport : IClientTransport, IAsyncDisposable
/// <inheritdoc />
public async Task<ITransport> ConnectAsync(CancellationToken cancellationToken = default)
{
if (_options.UseStreamableHttp)
{
return new StreamableHttpClientSessionTransport(_options, _httpClient, _loggerFactory, Name);
}
var sessionTransport = new SseClientSessionTransport(_options, _httpClient, _loggerFactory, Name);
try

9
src/ModelContextProtocol/Protocol/Transport/SseClientTransportOptions.cs
View file

@ -30,13 +30,20 @@ public record SseClientTransportOptions
}
}
/// <summary>
/// Gets or sets a value indicating whether to use "Streamable HTTP" for the transport rather than "HTTP with SSE". Defaults to false.
/// <see href="https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http">Streamable HTTP transport specification</see>.
/// <see href="https://modelcontextprotocol.io/specification/2024-11-05/basic/transports#http-with-sse">HTTP with SSE transport specification</see>.
/// </summary>
public bool UseStreamableHttp { get; init; }
/// <summary>
/// Gets a transport identifier used for logging purposes.
/// </summary>
public string? Name { get; init; }
/// <summary>
/// Gets or sets a timeout used to establish the initial connection to the SSE server.
/// Gets or sets a timeout used to establish the initial connection to the SSE server. Defaults to 30 seconds.
/// </summary>
/// <remarks>
/// This timeout controls how long the client waits for:

8
src/ModelContextProtocol/Protocol/Transport/StreamClientSessionTransport.cs
View file

@ -146,15 +146,7 @@ internal class StreamClientSessionTransport : TransportBase
var message = (JsonRpcMessage?)JsonSerializer.Deserialize(line.AsSpan().Trim(), McpJsonUtilities.DefaultOptions.GetTypeInfo(typeof(JsonRpcMessage)));
if (message != null)
{
string messageId = "(no id)";
if (message is JsonRpcMessageWithId messageWithId)
{
messageId = messageWithId.Id.ToString();
}
LogTransportReceivedMessage(Name, messageId);
await WriteMessageAsync(message, cancellationToken).ConfigureAwait(false);
LogTransportMessageWritten(Name, messageId);
}
else
{

8
src/ModelContextProtocol/Protocol/Transport/StreamServerTransport.cs
View file

@ -111,15 +111,7 @@ public class StreamServerTransport : TransportBase
{
if (JsonSerializer.Deserialize(line, McpJsonUtilities.DefaultOptions.GetTypeInfo(typeof(JsonRpcMessage))) is JsonRpcMessage message)
{
string messageId = "(no id)";
if (message is JsonRpcMessageWithId messageWithId)
{
messageId = messageWithId.Id.ToString();
}
LogTransportReceivedMessage(Name, messageId);
await WriteMessageAsync(message, shutdownToken).ConfigureAwait(false);
LogTransportMessageWritten(Name, messageId);
}
else
{

241
src/ModelContextProtocol/Protocol/Transport/StreamableHttpClientSessionTransport.cs Normal file
View file

@ -0,0 +1,241 @@
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ModelContextProtocol.Protocol.Messages;
using ModelContextProtocol.Utils;
using ModelContextProtocol.Utils.Json;
using System.Net.Http.Headers;
using System.Net.ServerSentEvents;
using System.Text.Json;
#if NET
using System.Net.Http.Json;
#else
using System.Text;
#endif
namespace ModelContextProtocol.Protocol.Transport;
/// <summary>
/// The Streamable HTTP client transport implementation
/// </summary>
internal sealed partial class StreamableHttpClientSessionTransport : TransportBase
{
private static readonly MediaTypeWithQualityHeaderValue s_applicationJsonMediaType = new("application/json");
private static readonly MediaTypeWithQualityHeaderValue s_textEventStreamMediaType = new("text/event-stream");
private readonly HttpClient _httpClient;
private readonly SseClientTransportOptions _options;
private readonly CancellationTokenSource _connectionCts;
private readonly ILogger _logger;
private string? _mcpSessionId;
private Task? _getReceiveTask;
public StreamableHttpClientSessionTransport(SseClientTransportOptions transportOptions, HttpClient httpClient, ILoggerFactory? loggerFactory, string endpointName)
: base(endpointName, loggerFactory)
{
Throw.IfNull(transportOptions);
Throw.IfNull(httpClient);
_options = transportOptions;
_httpClient = httpClient;
_connectionCts = new CancellationTokenSource();
_logger = (ILogger?)loggerFactory?.CreateLogger<SseClientTransport>() ?? NullLogger.Instance;
// We connect with the initialization request with the MCP transport. This means that any errors won't be observed
// until the first call to SendMessageAsync. Fortunately, that happens internally in McpClientFactory.ConnectAsync
// so we still throw any connection-related Exceptions from there and never expose a pre-connected client to the user.
SetConnected(true);
}
/// <inheritdoc/>
public override async Task SendMessageAsync(
JsonRpcMessage message,
CancellationToken cancellationToken = default)
{
using var sendCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _connectionCts.Token);
cancellationToken = sendCts.Token;
#if NET
using var content = JsonContent.Create(message, McpJsonUtilities.DefaultOptions.GetTypeInfo<JsonRpcMessage>());
#else
using var content = new StringContent(
JsonSerializer.Serialize(message, McpJsonUtilities.JsonContext.Default.JsonRpcMessage),
Encoding.UTF8,
"application/json"
);
#endif
using var httpRequestMessage = new HttpRequestMessage(HttpMethod.Post, _options.Endpoint)
{
Content = content,
Headers =
{
Accept = { s_applicationJsonMediaType, s_textEventStreamMediaType },
},
};
CopyAdditionalHeaders(httpRequestMessage.Headers, _options.AdditionalHeaders, _mcpSessionId);
using var response = await _httpClient.SendAsync(httpRequestMessage, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
response.EnsureSuccessStatusCode();
var rpcRequest = message as JsonRpcRequest;
JsonRpcMessage? rpcResponseCandidate = null;
if (response.Content.Headers.ContentType?.MediaType == "application/json")
{
var responseContent = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
rpcResponseCandidate = await ProcessMessageAsync(responseContent, cancellationToken).ConfigureAwait(false);
}
else if (response.Content.Headers.ContentType?.MediaType == "text/event-stream")
{
using var responseBodyStream = await response.Content.ReadAsStreamAsync(cancellationToken);
rpcResponseCandidate = await ProcessSseResponseAsync(responseBodyStream, rpcRequest, cancellationToken).ConfigureAwait(false);
}
if (rpcRequest is null)
{
return;
}
if (rpcResponseCandidate is not JsonRpcMessageWithId messageWithId || messageWithId.Id != rpcRequest.Id)
{
throw new McpException($"Streamable HTTP POST response completed without a reply to request with ID: {rpcRequest.Id}");
}
if (rpcRequest.Method == RequestMethods.Initialize && rpcResponseCandidate is JsonRpcResponse)
{
// We've successfully initialized! Copy session-id and start GET request if any.
if (response.Headers.TryGetValues("mcp-session-id", out var sessionIdValues))
{
_mcpSessionId = sessionIdValues.FirstOrDefault();
}
_getReceiveTask = ReceiveUnsolicitedMessagesAsync();
}
}
public override async ValueTask DisposeAsync()
{
try
{
await _connectionCts.CancelAsync().ConfigureAwait(false);
try
{
if (_getReceiveTask != null)
{
await _getReceiveTask.ConfigureAwait(false);
}
}
catch (OperationCanceledException)
{
}
finally
{
_connectionCts.Dispose();
}
}
finally
{
SetConnected(false);
}
}
private async Task ReceiveUnsolicitedMessagesAsync()
{
// Send a GET request to handle any unsolicited messages not sent over a POST response.
using var request = new HttpRequestMessage(HttpMethod.Get, _options.Endpoint);
request.Headers.Accept.Add(s_textEventStreamMediaType);
CopyAdditionalHeaders(request.Headers, _options.AdditionalHeaders, _mcpSessionId);
using var response = await _httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, _connectionCts.Token).ConfigureAwait(false);
if (!response.IsSuccessStatusCode)
{
// Server support for the GET request is optional. If it fails, we don't care. It just means we won't receive unsolicited messages.
return;
}
using var responseStream = await response.Content.ReadAsStreamAsync(_connectionCts.Token).ConfigureAwait(false);
await ProcessSseResponseAsync(responseStream, relatedRpcRequest: null, _connectionCts.Token).ConfigureAwait(false);
}
private async Task<JsonRpcMessageWithId?> ProcessSseResponseAsync(Stream responseStream, JsonRpcRequest? relatedRpcRequest, CancellationToken cancellationToken)
{
await foreach (SseItem<string> sseEvent in SseParser.Create(responseStream).EnumerateAsync(cancellationToken).ConfigureAwait(false))
{
if (sseEvent.EventType != "message")
{
continue;
}
var message = await ProcessMessageAsync(sseEvent.Data, cancellationToken).ConfigureAwait(false);
// The server SHOULD end the response here anyway, but we won't leave it to chance. This transport makes
// a GET request for any notifications that might need to be sent after the completion of each POST.
if (message is JsonRpcMessageWithId messageWithId && relatedRpcRequest?.Id == messageWithId.Id)
{
return messageWithId;
}
}
return null;
}
private async Task<JsonRpcMessage?> ProcessMessageAsync(string data, CancellationToken cancellationToken)
{
try
{
var message = JsonSerializer.Deserialize(data, McpJsonUtilities.JsonContext.Default.JsonRpcMessage);
if (message is null)
{
LogTransportMessageParseUnexpectedTypeSensitive(Name, data);
return null;
}
await WriteMessageAsync(message, cancellationToken).ConfigureAwait(false);
return message;
}
catch (JsonException ex)
{
LogJsonException(ex, data);
}
return null;
}
private void LogJsonException(JsonException ex, string data)
{
if (_logger.IsEnabled(LogLevel.Trace))
{
LogTransportMessageParseFailedSensitive(Name, data, ex);
}
else
{
LogTransportMessageParseFailed(Name, ex);
}
}
internal static void CopyAdditionalHeaders(HttpRequestHeaders headers, Dictionary<string, string>? additionalHeaders, string? sessionId = null)
{
if (sessionId is not null)
{
headers.Add("mcp-session-id", sessionId);
}
if (additionalHeaders is null)
{
return;
}
foreach (var header in additionalHeaders)
{
if (!headers.TryAddWithoutValidation(header.Key, header.Value))
{
throw new InvalidOperationException($"Failed to add header '{header.Key}' with value '{header.Value}' from {nameof(SseClientTransportOptions.AdditionalHeaders)}.");
}
}
}
}

2
src/ModelContextProtocol/Protocol/Transport/StreamableHttpPostTransport.cs
View file

@ -61,7 +61,7 @@ internal sealed class StreamableHttpPostTransport(ChannelWriter<JsonRpcMessage>?
{
yield return message;
if (message.Data is JsonRpcResponse response)
if (message.Data is JsonRpcMessageWithId response)
{
if (_pendingRequests.Remove(response.Id) && _pendingRequests.Count == 0)
{

11
src/ModelContextProtocol/Protocol/Transport/TransportBase.cs
View file

@ -37,7 +37,7 @@ public abstract partial class TransportBase : ITransport
_messageChannel = Channel.CreateUnbounded<JsonRpcMessage>(new UnboundedChannelOptions
{
SingleReader = true,
SingleWriter = true,
SingleWriter = false,
});
}
@ -76,6 +76,12 @@ public abstract partial class TransportBase : ITransport
throw new InvalidOperationException("Transport is not connected");
}
if (_logger.IsEnabled(LogLevel.Debug))
{
var messageId = (message as JsonRpcMessageWithId)?.Id.ToString() ?? "(no id)";
LogTransportReceivedMessage(Name, messageId);
}
await _messageChannel.Writer.WriteAsync(message, cancellationToken).ConfigureAwait(false);
}
@ -115,9 +121,6 @@ public abstract partial class TransportBase : ITransport
[LoggerMessage(Level = LogLevel.Debug, Message = "{EndpointName} transport received message with ID '{MessageId}'.")]
private protected partial void LogTransportReceivedMessage(string endpointName, string messageId);
[LoggerMessage(Level = LogLevel.Debug, Message = "{EndpointName} transport sent message with ID '{MessageId}'.")]
private protected partial void LogTransportMessageWritten(string endpointName, string messageId);
[LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} transport received unexpected message. Message: '{Message}'.")]
private protected partial void LogTransportMessageParseUnexpectedTypeSensitive(string endpointName, string message);