feat: client-side streamable-http transport supports resumability #380
Conversation
WalkthroughThis update introduces support for resuming interrupted Server-Sent Events (SSE) streams in the StreamableHTTP client by tracking the last received event ID and retrying failed connections. It adds a new resumption option, related methods, and updates SSE event handling to propagate event IDs. Minor whitespace cleanups are also included in test files. Changes
Possibly related PRs
Suggested labels
Suggested reviewers
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 golangci-lint (1.64.8)Error: you are using a configuration file for golangci-lint v2 with golangci-lint v1: please use golangci-lint v2 ✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
client/transport/streamable_http.go (2)
334-340: Improve retry loop readability and error message specificityThe retry logic is correct but could be more idiomatic and the error message more specific.
- for range c.maxRetryCount { - resp, err, canRetry := c.performSSEResumption(ctx, lastEventId, resumptionCallback) - if err == nil || lastEventId == "" || !canRetry { - return resp, err - } - } - return nil, fmt.Errorf("failed to retrieve response after attempting resumption %d times", c.maxRetryCount) + for i := 0; i < c.maxRetryCount; i++ { + resp, err, canRetry := c.performSSEResumption(ctx, lastEventId, resumptionCallback) + if err == nil || lastEventId == "" || !canRetry { + return resp, err + } + } + return nil, fmt.Errorf("failed to resume SSE stream after %d retry attempts", c.maxRetryCount)
57-437: Consider adding tests for the resumption functionalityThe resumption implementation is well-designed and follows the MCP specification. However, I notice that tests for this new functionality are not included in this PR. As mentioned in the PR objectives, tests are planned for a future commit.
Would you like me to help generate comprehensive test cases for the resumption functionality, including:
- Successful resumption scenarios
- Retry exhaustion cases
- Error handling during resumption
- Null byte filtering in event IDs
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
client/transport/sse_test.go(2 hunks)client/transport/streamable_http.go(10 hunks)client/transport/streamable_http_test.go(2 hunks)
🔇 Additional comments (6)
client/transport/streamable_http_test.go (1)
420-420: LGTM - Whitespace cleanupRemoving trailing blank lines improves code consistency.
Also applies to: 440-440
client/transport/sse_test.go (1)
411-411: LGTM - Consistent whitespace cleanupRemoving trailing blank lines to maintain consistency with the changes in
streamable_http_test.go.Also applies to: 413-413, 452-452
client/transport/streamable_http.go (4)
57-75: Well-implemented resumption optionThe
WithResumptionfunction properly validates input and follows the established pattern for configuration options. Good documentation referencing the MCP specification.
90-96: Appropriate struct fields for resumption supportThe new fields
resumptionEnabledandmaxRetryCountare well-placed and properly support the resumption feature.
346-437: Excellent implementation of SSE resumption logicThe
performSSEResumptionmethod is comprehensive and handles all edge cases properly:
- Correct use of GET request with
Last-Event-Idheader- Appropriate error handling with meaningful
canRetryflag- Consistent OAuth and session management
559-566: Good security consideration for event ID handlingExcellent implementation that properly filters null bytes to prevent HTTP header injection vulnerabilities. The comment clearly explains the security rationale.
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
client/transport/streamable_http.go (1)
372-374: Validate empty lastEventId parameter handling.The check for empty
lastEventIdis good, but this condition should theoretically never occur based on the calling logic inSendRequest. Consider adding a more descriptive error message or debug logging.- if lastEventId == "" { - return nil, fmt.Errorf("sse resumption request requires a last event id"), false - } + if lastEventId == "" { + return nil, fmt.Errorf("sse resumption request requires a last event id (this should not happen)"), false + }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
client/transport/streamable_http.go(10 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (1)
client/transport/streamable_http.go (3)
client/transport/interface.go (2)
HTTPHeaderFunc(13-13)JSONRPCResponse(41-50)mcp/types.go (1)
JSONRPCResponse(324-328)client/oauth.go (1)
OAuthAuthorizationRequiredError(61-61)
🔇 Additional comments (9)
client/transport/streamable_http.go (9)
57-74: Well-implemented resumption configuration option.The function properly validates the retry count and provides clear documentation referencing the MCP specification. The parameter validation ensuring
maxRetryCount >= 1prevents invalid configurations.
90-95: Struct fields properly integrated for resumption support.The new fields
resumptionEnabledandmaxRetryCountare appropriately added to support the resumption functionality without breaking existing code.
181-182: Header constants follow existing patterns.The new
headerKeyLastEventIDconstant is properly defined following the same naming convention as the existingheaderKeySessionID.
346-439: Comprehensive resumption method with proper error handling.The
performSSEResumptionmethod is well-implemented with appropriate OAuth handling, session management, and error classification via thecanRetryboolean return value.
443-447: Method signature change is appropriate for resumption support.The addition of the
resumptionCallbackparameter maintains backward compatibility since this is an internal method and enables the resumption functionality.
470-472: Event ID tracking correctly integrated.The resumption callback is properly invoked when an event ID is present, enabling the retry mechanism to track progress through the SSE stream.
561-567: Excellent security practice with null byte filtering.The null byte filtering prevents HTTP header injection attacks when the event ID is sent back in the
Last-Event-Idheader. This is a crucial security consideration.
507-571: Enhanced SSE parsing with proper event ID handling.The modifications to
readSSEcorrectly parse theid:field from SSE streams and properly reset the ID after each event. The implementation follows the SSE specification accurately.
323-340:✅ Verification successful
Verify retry loop termination conditions.
The resumption logic looks correct, but ensure the loop termination is robust. The current implementation could potentially retry on non-recoverable errors.
Consider adding exponential backoff or different retry strategies:
- for range c.maxRetryCount { + for attempt := 1; attempt <= c.maxRetryCount; attempt++ { + if attempt > 1 { + // Add exponential backoff delay + time.Sleep(time.Duration(attempt) * 100 * time.Millisecond) + } resp, err, canRetry := c.performSSEResumption(ctx, lastEventId, resumptionCallback)
🏁 Script executed:
#!/bin/bash # Search for similar retry patterns in the codebase to ensure consistency rg -A 5 -B 5 "for.*range.*maxRetry" --type goLength of output: 781
🏁 Script executed:
#!/bin/bash # Locate the declaration of maxRetryCount in client/transport/streamable_http.go rg -n "maxRetryCount" client/transport/streamable_http.goLength of output: 385
Retry loop termination verified—no infinite retries
The loop runs exactlyc.maxRetryCounttimes and exits early on success (err == nil), missing IDs (lastEventId == ""), or non-recoverable errors (canRetry == false). No change is strictly required.Optional improvement: switch to an index-based loop to add exponential backoff:
- for range c.maxRetryCount { + for attempt := 1; attempt <= c.maxRetryCount; attempt++ { + if attempt > 1 { + time.Sleep(time.Duration(attempt) * 100 * time.Millisecond) + } resp, err, canRetry := c.performSSEResumption(ctx, lastEventId, resumptionCallback) if err == nil || lastEventId == "" || !canRetry { return resp, err } }
Description
Implemented for client-side streamable-http transport
Fixes #<issue_number> (if applicable)
Type of Change
Checklist
MCP Spec Compliance
Additional Information
Will add the tests in a later commit
Summary by CodeRabbit
New Features
Style