forked from gdsoumya/protomcp
-
Notifications
You must be signed in to change notification settings - Fork 0
Add stateless subscriptions example with listen-based e2e #9
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Closed
Closed
Changes from 4 commits
Commits
Show all changes
9 commits
Select commit
Hold shift + click to select a range
168268f
feat(examples): add stateless subscriptions example with listen-based…
yhxlele 0ac09f3
docs: document the stateless serving shape in the subscriptions READMEs
yhxlele 2b95c26
docs: drop the false claim that a dropped listen stream is re-issued …
yhxlele 760c89e
docs: re-reads are a reconciliation fallback, not a liveness signal f…
yhxlele 2b1b8ed
docs: liveness must ride each watched URI's stream; recovery is sessi…
yhxlele 3e9375c
feat(examples): implement the per-URI heartbeat the stateless README …
yhxlele facdbd8
fix(examples): reject request-scoped resources/subscribe on the state…
yhxlele 049b58e
fix: guard legacy stateless subscriptions
jiachengxu 189caf2
chore(lint): whitelist 'cancelled' (MCP's own spelling) and restore t…
yhxlele File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Some comments aren't visible on the classic Files Changed page.
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
189 changes: 189 additions & 0 deletions
189
examples/subscriptions/cmd/subscriptions-stateless/main.go
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,189 @@ | ||
| // Command subscriptions-stateless runs the tasks-mcp server in | ||
| // stateless Streamable HTTP mode — the deployment shape for | ||
| // horizontally scaled servers behind a plain round-robin load | ||
| // balancer, with no session affinity anywhere. | ||
| // | ||
| // Stateless mode is also the only mode in which the Go SDK speaks | ||
| // protocol revision 2026-07-28, which replaces `resources/subscribe` | ||
| // with `subscriptions/listen`: one long-lived POST whose response | ||
| // stream carries the notifications. That flips where subscription | ||
| // state lives — on the connection, not in a server-side session map: | ||
| // | ||
| // - Any replica can serve any request; nothing routes on a session. | ||
| // - A subscription lives exactly as long as its listen stream. The | ||
| // replica holding the stream delivers `notifications/resources/ | ||
| // updated` for it. A dropped stream is NOT re-issued: on go-sdk | ||
| // v1.7.0 these streams carry no SSE event IDs, the client abandons | ||
| // a listen POST that dies without one, and — because the listen | ||
| // call is dispatched fire-and-forget — no error surfaces to the | ||
| // application. A later Subscribe for the same URI is a no-op while | ||
| // the client still thinks it is subscribed; recovery is | ||
| // Unsubscribe then a fresh Subscribe, which any replica can | ||
| // answer. Clients that must survive drops need their own liveness | ||
| // signal (e.g. a subscribed heartbeat resource) to notice a dead | ||
| // stream. | ||
| // - SubscribeHandler / UnsubscribeHandler fire per URI exactly as in | ||
| // the legacy modes — `subscriptions/listen` routes through the same | ||
| // gate — so ACL checks carry over unchanged. | ||
| // - Each replica pushes ResourceUpdated for events it observes. With | ||
| // a shared event source (pub/sub, CDC, PG LISTEN) every replica | ||
| // sees every event, so whichever replica holds a given listen | ||
| // stream delivers to it. | ||
| // | ||
| // PropagateRequestCancellation ties every in-flight handler's context | ||
| // to its HTTP request: when the client goes away mid-call, the handler | ||
| // is canceled instead of running to completion for nobody. (For | ||
| // subscriptions/listen the SDK forces this on regardless — a listen | ||
| // handler blocks until its request ends, so it would otherwise never | ||
| // return.) | ||
| // | ||
| // Usage: | ||
| // | ||
| // go run ./examples/subscriptions/cmd/subscriptions-stateless # listens on 127.0.0.1:8080 | ||
| // go run ./examples/subscriptions/cmd/subscriptions-stateless -addr :9000 | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "flag" | ||
| "fmt" | ||
| "log" | ||
| "net" | ||
| "net/http" | ||
| "os" | ||
| "os/signal" | ||
| "syscall" | ||
| "time" | ||
|
|
||
| "github.com/modelcontextprotocol/go-sdk/mcp" | ||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/credentials/insecure" | ||
|
|
||
| tasksserver "github.com/akuity/protomcp/examples/tasks/server" | ||
| tasksv1 "github.com/akuity/protomcp/pkg/api/gen/examples/tasks/v1" | ||
| "github.com/akuity/protomcp/pkg/protomcp" | ||
| ) | ||
|
|
||
| func main() { | ||
| addr := flag.String("addr", "127.0.0.1:8080", "HTTP listen address for the MCP server") | ||
| flag.Parse() | ||
|
|
||
| ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) | ||
| err := run(ctx, *addr) | ||
| stop() | ||
| if err != nil { | ||
| log.Fatalf("subscriptions-stateless: %v", err) | ||
| } | ||
| } | ||
|
|
||
| // newStatelessServer builds the MCP server in the shape this example | ||
| // exists to demonstrate. Split out so the e2e tests exercise exactly | ||
| // what the binary runs; nil handlers default to allow-all. The SDK | ||
| // calls the subscribe/unsubscribe gate per URI on both the legacy | ||
| // resources/subscribe path and the 2026-07-28 subscriptions/listen | ||
| // path, so this is where an ACL would live. | ||
| func newStatelessServer( | ||
| grpcClient tasksv1.TasksClient, | ||
| onSubscribe func(context.Context, *mcp.SubscribeRequest) error, | ||
| onUnsubscribe func(context.Context, *mcp.UnsubscribeRequest) error, | ||
| ) *protomcp.Server { | ||
| if onSubscribe == nil { | ||
| onSubscribe = func(context.Context, *mcp.SubscribeRequest) error { return nil } | ||
| } | ||
| if onUnsubscribe == nil { | ||
| onUnsubscribe = func(context.Context, *mcp.UnsubscribeRequest) error { return nil } | ||
| } | ||
| srv := protomcp.New("tasks-subscriptions-stateless-mcp", "0.1.0", | ||
| protomcp.WithSDKOptions(&mcp.ServerOptions{ | ||
| SubscribeHandler: onSubscribe, | ||
| UnsubscribeHandler: onUnsubscribe, | ||
| }), | ||
| protomcp.WithHTTPOptions(&mcp.StreamableHTTPOptions{ | ||
| Stateless: true, | ||
| PropagateRequestCancellation: true, | ||
| }), | ||
| ) | ||
| tasksv1.RegisterTasksMCPTools(srv, grpcClient) | ||
| tasksv1.RegisterTasksMCPResources(srv, grpcClient) | ||
| return srv | ||
| } | ||
|
|
||
| func run(ctx context.Context, addr string) error { | ||
| // 1. Start the Tasks gRPC service. Its OnChange hook fires on every | ||
| // CRUD mutation and becomes our push point below. In a real | ||
| // multi-replica deployment this would be a shared event source | ||
| // (pub/sub, CDC, PG LISTEN) consumed by every replica. | ||
| tSrv := tasksserver.New() | ||
| grpcClient, shutdownGRPC, err := startTasksGRPC(ctx, tSrv) | ||
| if err != nil { | ||
| return fmt.Errorf("start grpc: %w", err) | ||
| } | ||
| defer shutdownGRPC() | ||
|
|
||
| // 2. Build the stateless MCP server. | ||
| srv := newStatelessServer(grpcClient, nil, nil) | ||
|
|
||
| // 3. Push path: identical to the stateful examples. The SDK routes | ||
| // each ResourceUpdated to whichever live listen streams (or | ||
| // legacy sessions) subscribed to that URI on this replica. | ||
| tSrv.OnChange = func(id string) { | ||
| uri := "tasks://" + id | ||
| if nErr := srv.SDK().ResourceUpdated(ctx, &mcp.ResourceUpdatedNotificationParams{URI: uri}); nErr != nil { | ||
| log.Printf("ResourceUpdated %s: %v", uri, nErr) | ||
| } | ||
| } | ||
|
|
||
| httpSrv := &http.Server{ | ||
| Addr: addr, | ||
| Handler: srv, | ||
| ReadHeaderTimeout: 5 * time.Second, | ||
| } | ||
|
|
||
| fmt.Printf("tasks-subscriptions-stateless-mcp listening on %s (stateless, protocol >= 2026-07-28 capable)\n", addr) | ||
| fmt.Println(" resources: tasks://{id} (read + list + push-on-mutation subscribe via subscriptions/listen)") | ||
| fmt.Println(" tools: Tasks_ListTasks, Tasks_GetTask, Tasks_CreateTask,") | ||
| fmt.Println(" Tasks_UpdateTask, Tasks_DeleteTask") | ||
|
|
||
| errCh := make(chan error, 1) | ||
| go func() { | ||
| if sErr := httpSrv.ListenAndServe(); sErr != nil && !errors.Is(sErr, http.ErrServerClosed) { | ||
| errCh <- sErr | ||
| return | ||
| } | ||
| errCh <- nil | ||
| }() | ||
|
|
||
| select { | ||
| case <-ctx.Done(): | ||
| shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | ||
| defer cancel() | ||
| return httpSrv.Shutdown(shutdownCtx) | ||
| case sErr := <-errCh: | ||
| return sErr | ||
| } | ||
| } | ||
|
|
||
| func startTasksGRPC(ctx context.Context, impl tasksv1.TasksServer) (tasksv1.TasksClient, func(), error) { | ||
| lis, err := (&net.ListenConfig{}).Listen(ctx, "tcp", "127.0.0.1:0") | ||
| if err != nil { | ||
| return nil, nil, fmt.Errorf("listen: %w", err) | ||
| } | ||
| grpcSrv := grpc.NewServer() | ||
| tasksv1.RegisterTasksServer(grpcSrv, impl) | ||
| go func() { _ = grpcSrv.Serve(lis) }() | ||
|
|
||
| conn, err := grpc.NewClient(lis.Addr().String(), | ||
| grpc.WithTransportCredentials(insecure.NewCredentials())) | ||
| if err != nil { | ||
| grpcSrv.Stop() | ||
| _ = lis.Close() | ||
| return nil, nil, fmt.Errorf("dial: %w", err) | ||
| } | ||
|
|
||
| cleanup := func() { | ||
| _ = conn.Close() | ||
| grpcSrv.GracefulStop() | ||
| } | ||
| return tasksv1.NewTasksClient(conn), cleanup, nil | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think the heartbeat has to be sent on each watched URI's own stream, since
ClientSession.Subscribeopens a separatesubscriptions/listenPOST for each URI, and a separate heartbeat subscription can keep arriving while watched URI's stream is already dead. Could we recommend either emitting a periodic update for every watched URI, or unconditionallyUnsubscribe+Subscribeeach URI, and update the same guidance inmain.go?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Right about the per-URI streams —
Subscribeopens a dedicatedsubscriptions/listenPOST per URI (client.go:1375, and theresourceSubsdoc comment says "dedicated" outright), so a separate heartbeat subscription attests only itself. But pinning the recommended cycle in an e2e surfaced something worse: on v1.7.0 the Unsubscribe + Subscribe cycle can't work at all. Unsubscribe cancels the listen call, the client then sendsnotifications/cancelledwithout the SEP-2575_metaenvelope (cancelCallis the one send site that skipsinjectRequestMeta), a conformant server rejects it (-32602/ HTTP 400), and the client treats the failed send as fatal and hard-fails the whole connection — the re-Subscribe is dispatched fire-and-forget into the dead conn and vanishes silently, and canceling any in-flight call poisons the session the same way. Filed upstream as modelcontextprotocol/go-sdk#1212 with fix PR modelcontextprotocol/go-sdk#1213.Guidance updated accordingly in 2b1b8ed: detection = the server emits a periodic update for every watched URI, so a missed beat on that URI's own stream marks exactly that stream dead; recovery = replace the session (Close + Connect + re-Subscribe each URI + re-read to reconcile) until the SDK fix lands. 3e9375c then implements the heartbeat in this example rather than just recommending it —
watchHeartbeatrefcounts watched URIs from the subscribe/unsubscribe gate and touches each on the-heartbeatinterval — somain.goand the README say the same thing. The e2e now pins all three pieces: heartbeats arriving with no mutation at all, the Unsubscribe poisoning, and the reconnect recovery; the poisoning assertions flip to failing when go-sdk#1212 is fixed, flagging this guidance for revisit (the cheaper per-URI cycle becomes viable then).There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Follow-up: release-day adoption is tracked on our side — the go-sdk pin bump flips this PR's poisoning e2e (it asserts the broken behavior by design) and restores per-URI Unsubscribe + Subscribe cycling as the recommended recovery; #8's replay skip is version-locked to v1.7.0, so the same bump turns it into a hard assertion too.