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

Add StreamableHttpHandler and WithHttpTransport (#291)

Browse source 
* Simplify ModelContextProtocol.AspNetCore README

* Add StreamableHttpHandler

* Use "docker info" in CheckIsDockerAvailable

* Call WithHttpTransport in samples, tests and README

* Cleanup test namespaces

* Add CanConnect_WithMcpClient_AfterCustomizingRoute test

* Simplify relative URI handling

* Handle request made directly to the MapMcp route pattern

* Add Messages_FromNewUser_AreRejected test

* Fix README

* Shorten UserIdClaim ValueTuple names

* Remove MaxReconnectAttempts and ReconnectDelay from SseClientTransportOptions

- Add proper AdditionalHeaders support
This commit is contained in:
Stephen Halter 2025-04-14 07:15:33 -07:00 • committed by GitHub
commit 9c124b1770
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 597 additions and 324 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -101,11 +101,12 @@ internal sealed class SseClientSessionTransport : TransportBase
messageId = messageWithId.Id.ToString();
}
var response = await _httpClient.PostAsync(
_messageEndpoint,
content,
cancellationToken
).ConfigureAwait(false);
var httpRequestMessage = new HttpRequestMessage(HttpMethod.Post, _messageEndpoint)
{
Content = content,
};
CopyAdditionalHeaders(httpRequestMessage.Headers);
var response = await _httpClient.SendAsync(httpRequestMessage, cancellationToken).ConfigureAwait(false);
response.EnsureSuccessStatusCode();
@ -182,72 +183,52 @@ internal sealed class SseClientSessionTransport : TransportBase
private async Task ReceiveMessagesAsync(CancellationToken cancellationToken)
{
int reconnectAttempts = 0;
while (!cancellationToken.IsCancellationRequested && !IsConnected)
try
{
try
{
using var request = new HttpRequestMessage(HttpMethod.Get, _sseEndpoint);
request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
using var request = new HttpRequestMessage(HttpMethod.Get, _sseEndpoint);
request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
CopyAdditionalHeaders(request.Headers);
if (_options.AdditionalHeaders != null)
using var response = await _httpClient.SendAsync(
request,
HttpCompletionOption.ResponseHeadersRead,
cancellationToken
).ConfigureAwait(false);
response.EnsureSuccessStatusCode();
using var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
await foreach (SseItem<string> sseEvent in SseParser.Create(stream).EnumerateAsync(cancellationToken).ConfigureAwait(false))
{
switch (sseEvent.EventType)
{
foreach (var header in _options.AdditionalHeaders)
{
request.Headers.Add(header.Key, header.Value);
}
case "endpoint":
HandleEndpointEvent(sseEvent.Data);
break;
case "message":
await ProcessSseMessage(sseEvent.Data, cancellationToken).ConfigureAwait(false);
break;
}
using var response = await _httpClient.SendAsync(
request,
HttpCompletionOption.ResponseHeadersRead,
cancellationToken
).ConfigureAwait(false);
response.EnsureSuccessStatusCode();
using var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
await foreach (SseItem<string> sseEvent in SseParser.Create(stream).EnumerateAsync(cancellationToken).ConfigureAwait(false))
{
switch (sseEvent.EventType)
{
case "endpoint":
HandleEndpointEvent(sseEvent.Data);
break;
case "message":
await ProcessSseMessage(sseEvent.Data, cancellationToken).ConfigureAwait(false);
break;
}
}
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
_logger.TransportReadMessagesCancelled(_endpointName);
// Normal shutdown
}
catch (IOException) when (cancellationToken.IsCancellationRequested)
{
_logger.TransportReadMessagesCancelled(_endpointName);
// Normal shutdown
}
catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
{
_logger.TransportConnectionError(_endpointName, ex);
reconnectAttempts++;
if (reconnectAttempts >= _options.MaxReconnectAttempts)
{
throw new McpTransportException("Exceeded reconnect limit", ex);
}
await Task.Delay(_options.ReconnectDelay, cancellationToken).ConfigureAwait(false);
}
}
SetConnected(false);
catch when (cancellationToken.IsCancellationRequested)
{
// Normal shutdown
_connectionEstablished.TrySetCanceled(cancellationToken);
_logger.TransportReadMessagesCancelled(_endpointName);
}
catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
{
_connectionEstablished.TrySetException(ex);
_logger.TransportConnectionError(_endpointName, ex);
throw;
}
finally
{
SetConnected(false);
}
}
private async Task ProcessSseMessage(string data, CancellationToken cancellationToken)
@ -293,21 +274,8 @@ internal sealed class SseClientSessionTransport : TransportBase
return;
}
// Check if data is absolute URI
if (data.StartsWith("http://", StringComparison.OrdinalIgnoreCase) || data.StartsWith("https://", StringComparison.OrdinalIgnoreCase))
{
// Since the endpoint is an absolute URI, we can use it directly
_messageEndpoint = new Uri(data);
}
else
{
// If the endpoint is a relative URI, we need to combine it with the relative path of the SSE endpoint
var baseUriBuilder = new UriBuilder(_sseEndpoint);
// Instead of manually concatenating strings, use the Uri class's composition capabilities
_messageEndpoint = new Uri(baseUriBuilder.Uri, data);
}
// If data is an absolute URL, the Uri will be constructed entirely from it and not the _sseEndpoint.
_messageEndpoint = new Uri(_sseEndpoint, data);
// Set connected state
SetConnected(true);
@ -319,4 +287,18 @@ internal sealed class SseClientSessionTransport : TransportBase
throw new McpTransportException("Failed to parse endpoint event", ex);
}
}
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)}.");
}
}
}
}
}

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

@ -48,37 +48,6 @@ public record SseClientTransportOptions
/// </remarks>
public TimeSpan ConnectionTimeout { get; init; } = TimeSpan.FromSeconds(30);
/// <summary>
/// Gets or sets the maximum number of reconnection attempts for the SSE connection before giving up.
/// </summary>
/// <remarks>
/// <para>
/// This property controls how many times the client will attempt to reconnect to the SSE server
/// after a connection failure occurs. If all reconnection attempts fail, a
/// <see cref="McpTransportException"/> with the message "Exceeded reconnect limit" will be thrown.
/// </para>
/// <para>
/// Between each reconnection attempt, the client will wait for the duration specified by <see cref="ReconnectDelay"/>.
/// </para>
/// </remarks>
public int MaxReconnectAttempts { get; init; } = 3;
/// <summary>
/// Gets or sets the delay to employ between reconnection attempts when the SSE connection fails.
/// </summary>
/// <remarks>
/// <para>
/// When a connection to the SSE server is lost or fails, the client will wait for this duration
/// before attempting to reconnect. This helps prevent excessive reconnection attempts in quick succession
/// which could overload the server or network.
/// </para>
/// <para>
/// The reconnection process continues until either a successful connection is established or
/// the maximum number of reconnection attempts (<see cref="MaxReconnectAttempts"/>) is reached.
/// </para>
/// </remarks>
public TimeSpan ReconnectDelay { get; init; } = TimeSpan.FromSeconds(5);
/// <summary>
/// Gets custom HTTP headers to include in requests to the SSE server.
/// </summary>