diff --git a/README.md b/README.md index 0c0f1d8..588f4b0 100644 --- a/README.md +++ b/README.md @@ -381,7 +381,7 @@ There are two common shapes: 2. **Wrap an external source.** Real handlers that start/stop upstream delivery per subscription (open a gRPC stream, run a PG `LISTEN`, subscribe to a Redis/Kafka topic, register a webhook). Best when events come from outside the process. Still call `ResourceUpdated` on each incoming event. -[`examples/subscriptions`](examples/subscriptions) ships both as runnable demos: `cmd/subscriptions-simple` for the push-from-write-path pattern (about 30 lines of wiring) and `cmd/subscriptions` for the external-source pattern, with per-principal subscribe authorization and a watcher wrapped in `protomcp.RetryLoop`. +[`examples/subscriptions`](examples/subscriptions) ships all of this as runnable demos: `cmd/subscriptions-simple` for the push-from-write-path pattern (about 30 lines of wiring), `cmd/subscriptions` for the external-source pattern with per-principal subscribe authorization and a watcher wrapped in `protomcp.RetryLoop`, and `cmd/subscriptions-stateless` for the stateless serving shape where ≥ 2026-07-28 clients subscribe via `subscriptions/listen` with no session affinity anywhere. ### `protomcp.v1.prompt`, method option @@ -460,7 +460,7 @@ Each example is standalone, runnable, and has its own README. |---|---| | [`examples/greeter`](examples/greeter) | Tool primitive surface, unary + server-streaming RPCs, progress notifications with monotonic counter, **progress-token gRPC-metadata propagation**, `ToolErrorHandler`, `ToolResultProcessor` redaction, `ToolMiddleware` request mutation, SDK options pass-through, **`field_schema.exclude` schema masking round-trip** | | [`examples/tasks`](examples/tasks) | **Every declarative MCP primitive end-to-end.** Tools with `read_only` / `idempotent` / `destructive` hints + `OUTPUT_ONLY` stripping, **two `resource_template` annotations (`tasks://{id}`, `tags://{id}`)**, **a single `resource_list` that enumerates both types via `{type}://{id}` with `OffsetPagination`**, **prompts (`tasks_review`)**, **elicitation (confirm `DeleteTask`)**, plus `@example` markers and `enumDescriptions` on `TaskStatus` | -| [`examples/subscriptions`](examples/subscriptions) | **User-wired resource subscriptions** on top of the Tasks resource template. Per-principal subscribe authorization in `SubscribeHandler`/`UnsubscribeHandler`, plus a watcher wrapped in `protomcp.RetryLoop` pushing `ResourceUpdated`. Race-tested. | +| [`examples/subscriptions`](examples/subscriptions) | **User-wired resource subscriptions** on top of the Tasks resource template. Per-principal subscribe authorization in `SubscribeHandler`/`UnsubscribeHandler`, plus a watcher wrapped in `protomcp.RetryLoop` pushing `ResourceUpdated`. Also the **stateless serving shape** (`Stateless` + `PropagateRequestCancellation`): protocol ≥ 2026-07-28 with `subscriptions/listen`-delivered subscriptions and no session affinity. Race-tested. | | [`examples/auth`](examples/auth) | Two-layer auth: SDK-native bearer middleware **or** custom HTTP middleware, both writing gRPC metadata for the upstream | Cmd directories inside each example hold the runnable binaries: @@ -473,6 +473,7 @@ Cmd directories inside each example hold the runnable binaries: - [`examples/greeter/cmd/sdkopts`](examples/greeter/cmd/sdkopts), pass `mcp.ServerOptions` / `mcp.StreamableHTTPOptions` - [`examples/tasks/cmd/tasks`](examples/tasks/cmd/tasks), CRUD - [`examples/subscriptions/cmd/subscriptions`](examples/subscriptions/cmd/subscriptions), authorization-gated subscribe wiring +- [`examples/subscriptions/cmd/subscriptions-stateless`](examples/subscriptions/cmd/subscriptions-stateless), stateless serving + `subscriptions/listen` delivery (protocol ≥ 2026-07-28) - [`examples/auth/cmd/auth`](examples/auth/cmd/auth), custom HTTP middleware → ctx → metadata - [`examples/auth/cmd/sdkauth`](examples/auth/cmd/sdkauth), MCP Go SDK's `auth.RequireBearerToken` → `TokenInfoFromContext` → metadata diff --git a/examples/subscriptions/README.md b/examples/subscriptions/README.md index 1683922..8cae837 100644 --- a/examples/subscriptions/README.md +++ b/examples/subscriptions/README.md @@ -195,6 +195,63 @@ resources push from internal code and others need an external watch. Return `nil` for push-path URIs (no-op) and open the external watch only for URIs that need it. +## Serving stateless (protocol ≥ 2026-07-28, no session affinity) + +`cmd/subscriptions-stateless` runs the same subscription wiring with + +```go +protomcp.WithHTTPOptions(&mcp.StreamableHTTPOptions{ + Stateless: true, + PropagateRequestCancellation: true, +}) +``` + +Stateless mode is the shape for horizontally scaled servers behind a +plain round-robin load balancer — and it is the only mode in which the +Go SDK speaks protocol revision 2026-07-28, where `resources/subscribe` +is replaced by `subscriptions/listen`: one long-lived POST whose +response stream carries the notifications. Subscription state lives on +that connection, not in a server-side session map, so no affinity +mechanism is needed: any replica can serve any request, and the replica +holding a listen stream delivers to it. Your `SubscribeHandler` / +`UnsubscribeHandler` fire per URI exactly as in the other patterns — +`subscriptions/listen` routes through the same gate — and the push +side (`ResourceUpdated`) is unchanged; in a multi-replica deployment, +feed every replica from a shared event source so whichever one holds a +given stream can deliver. + +**A dropped listen stream is not replaced, and the loss is silent.** On +go-sdk v1.7.0, streams on this protocol carry no SSE event IDs, and the +client abandons a POST stream whose connection dies without one instead +of retrying it; because `subscriptions/listen` is dispatched +fire-and-forget, the synthesized `request terminated without response` +error is discarded, so no error surfaces to the application. A later +`ClientSession.Subscribe` for the same URI is a no-op while the client +still believes it is subscribed — recovery is `Unsubscribe` (which +clears that client-side entry) followed by a fresh `Subscribe`, and any +replica can answer the new stream. A client that must survive +connection drops therefore needs a liveness signal that rides the +stream itself: a subscribed heartbeat resource the server touches on an +interval, where a missed-heartbeat timeout marks the stream dead and +triggers the recovery above — or, more bluntly, an unconditional +periodic re-subscribe. Periodic re-reads of the watched resource are a +reconciliation fallback, not a liveness check: each read is its own +stateless POST and succeeds whether or not the listen stream is alive, +so polling bounds how stale a client can silently become, but an +unchanged resource reveals nothing and a dead stream goes undetected. + +`PropagateRequestCancellation` ties each in-flight handler context to +its HTTP request, so a client that goes away mid-call cancels the +handler instead of leaving it running for nobody. (The SDK forces this +on for `subscriptions/listen` requests regardless — a listen handler +blocks until its request ends.) + +The e2e suite in `cmd/subscriptions-stateless/main_test.go` pins the +whole contract: 2026-07-28 negotiated over plain HTTP, listen-based +delivery to concurrent clients, unsubscribe teardown, handler +cancellation on request abort, and the same endpoint still answering a +classic `initialize` from pre-2026 clients. + ## Running the demos ```shell @@ -203,6 +260,10 @@ go run ./examples/subscriptions/cmd/subscriptions-simple -addr :8080 # Pattern B: watch stream + authz (requires a bearer token) go run ./examples/subscriptions/cmd/subscriptions -addr :8080 + +# Stateless: same push wiring, Stateless + PropagateRequestCancellation, +# subscriptions arrive via subscriptions/listen (protocol >= 2026-07-28) +go run ./examples/subscriptions/cmd/subscriptions-stateless -addr :8080 ``` Point any MCP client at `http://localhost:8080`. For Pattern B, diff --git a/examples/subscriptions/cmd/subscriptions-stateless/main.go b/examples/subscriptions/cmd/subscriptions-stateless/main.go new file mode 100644 index 0000000..9815dc9 --- /dev/null +++ b/examples/subscriptions/cmd/subscriptions-stateless/main.go @@ -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 +} diff --git a/examples/subscriptions/cmd/subscriptions-stateless/main_test.go b/examples/subscriptions/cmd/subscriptions-stateless/main_test.go new file mode 100644 index 0000000..d0f520d --- /dev/null +++ b/examples/subscriptions/cmd/subscriptions-stateless/main_test.go @@ -0,0 +1,355 @@ +// Wire-level e2e for the stateless serving shape: protocol 2026-07-28 +// negotiated over plain HTTP with no session affinity, resource +// subscriptions delivered over subscriptions/listen streams, handler +// cancellation tied to the HTTP request, and legacy clients still +// served by the same endpoint. +// +// Registration is asynchronous under subscriptions/listen (the client +// dispatches the listen call without awaiting), so delivery tests +// mutate in a poll loop until the notification arrives instead of +// mutating once and hoping the subscription was registered in time. +package main + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" + + 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" +) + +// harness bundles the running stateless server with the signal +// channels the tests observe. +type harness struct { + url string + grpcClient tasksv1.TasksClient + srv *protomcp.Server + subscribed chan string // URI per SubscribeHandler invocation + unsubscribed chan string // URI per UnsubscribeHandler invocation +} + +func startHarness(t *testing.T) *harness { + t.Helper() + tSrv := tasksserver.New() + grpcClient, shutdownGRPC, err := startTasksGRPC(context.Background(), tSrv) + if err != nil { + t.Fatalf("start grpc: %v", err) + } + t.Cleanup(shutdownGRPC) + + h := &harness{ + grpcClient: grpcClient, + subscribed: make(chan string, 16), + unsubscribed: make(chan string, 16), + } + h.srv = newStatelessServer(grpcClient, + func(_ context.Context, req *mcp.SubscribeRequest) error { + h.subscribed <- req.Params.URI + return nil + }, + func(_ context.Context, req *mcp.UnsubscribeRequest) error { + h.unsubscribed <- req.Params.URI + return nil + }, + ) + tSrv.OnChange = func(id string) { + _ = h.srv.SDK().ResourceUpdated(context.Background(), + &mcp.ResourceUpdatedNotificationParams{URI: "tasks://" + id}) + } + + ts := httptest.NewServer(h.srv) + t.Cleanup(ts.Close) + h.url = ts.URL + return h +} + +// connect dials the harness with the real SDK client over Streamable +// HTTP and asserts the session negotiated the modern protocol — the +// point of stateless serving. notif receives every ResourceUpdated URI. +func (h *harness) connect(ctx context.Context, t *testing.T, notif chan<- string) *mcp.ClientSession { + t.Helper() + opts := &mcp.ClientOptions{} + if notif != nil { + opts.ResourceUpdatedHandler = func(_ context.Context, req *mcp.ResourceUpdatedNotificationRequest) { + select { + case notif <- req.Params.URI: + default: + } + } + } + client := mcp.NewClient(&mcp.Implementation{Name: "stateless-test-client", Version: "0.0.1"}, opts) + cs, err := client.Connect(ctx, &mcp.StreamableClientTransport{Endpoint: h.url}, nil) + if err != nil { + t.Fatalf("client connect: %v", err) + } + t.Cleanup(func() { _ = cs.Close() }) + if v := cs.InitializeResult().ProtocolVersion; v < "2026-07-28" { + t.Fatalf("negotiated protocol %q; stateless serving must negotiate >= 2026-07-28 via server/discover", v) + } + return cs +} + +func (h *harness) createTask(ctx context.Context, t *testing.T, title string) *tasksv1.Task { + t.Helper() + task, err := h.grpcClient.CreateTask(ctx, &tasksv1.CreateTaskRequest{Task: &tasksv1.Task{Title: title}}) + if err != nil { + t.Fatalf("CreateTask: %v", err) + } + return task +} + +// mutateUntilNotified updates the task in a poll loop until notif +// yields its URI, absorbing the async listen registration. +func (h *harness) mutateUntilNotified(ctx context.Context, t *testing.T, task *tasksv1.Task, notif <-chan string) { + t.Helper() + wantURI := "tasks://" + task.Id + deadline := time.After(5 * time.Second) + tick := time.NewTicker(150 * time.Millisecond) + defer tick.Stop() + for { + select { + case uri := <-notif: + if uri != wantURI { + t.Fatalf("notified for %q, want %q", uri, wantURI) + } + return + case <-tick.C: + task.Title += "." + if _, err := h.grpcClient.UpdateTask(ctx, &tasksv1.UpdateTaskRequest{ + Id: task.Id, Title: task.Title, Description: task.Description, Done: task.Done, + }); err != nil { + t.Fatalf("UpdateTask: %v", err) + } + case <-deadline: + t.Fatalf("no resources/updated notification for %q within 5s", wantURI) + } + } +} + +func waitForURI(t *testing.T, ch <-chan string, want, what string) { + t.Helper() + select { + case uri := <-ch: + if uri != want { + t.Fatalf("%s fired for %q, want %q", what, uri, want) + } + case <-time.After(5 * time.Second): + t.Fatalf("%s did not fire for %q within 5s", what, want) + } +} + +// TestStateless_ToolCallsOnModernProtocol pins the baseline: a v1.7.0 +// client against the stateless endpoint negotiates >= 2026-07-28 (the +// connect helper asserts it) and normal tool round-trips work. +func TestStateless_ToolCallsOnModernProtocol(t *testing.T) { + ctx := context.Background() + h := startHarness(t) + cs := h.connect(ctx, t, nil) + + out, err := cs.CallTool(ctx, &mcp.CallToolParams{ + Name: "Tasks_CreateTask", + Arguments: json.RawMessage(`{"task":{"title":"stateless","done":false}}`), + }) + if err != nil { + t.Fatalf("CallTool: %v", err) + } + if out.IsError { + t.Fatalf("CallTool: unexpected IsError result: %+v", out) + } +} + +// TestStateless_SubscribeDeliversOverListen is the headline: a +// subscription made on a stateless server (where resources/subscribe +// does not exist) flows through SubscribeHandler and delivers +// resources/updated notifications over the subscriptions/listen +// stream. A second concurrent client on the same URI proves per- +// connection fan-out. +func TestStateless_SubscribeDeliversOverListen(t *testing.T) { + ctx := context.Background() + h := startHarness(t) + + notifA := make(chan string, 16) + notifB := make(chan string, 16) + csA := h.connect(ctx, t, notifA) + csB := h.connect(ctx, t, notifB) + + task := h.createTask(ctx, t, "watched") + uri := "tasks://" + task.Id + + if err := csA.Subscribe(ctx, &mcp.SubscribeParams{URI: uri}); err != nil { + t.Fatalf("Subscribe A: %v", err) + } + waitForURI(t, h.subscribed, uri, "SubscribeHandler (A)") + if err := csB.Subscribe(ctx, &mcp.SubscribeParams{URI: uri}); err != nil { + t.Fatalf("Subscribe B: %v", err) + } + waitForURI(t, h.subscribed, uri, "SubscribeHandler (B)") + + h.mutateUntilNotified(ctx, t, task, notifA) + // B holds its own listen stream; the same mutation burst must have + // reached it too (allow a beat for independent delivery). + select { + case gotB := <-notifB: + if gotB != uri { + t.Fatalf("client B notified for %q, want %q", gotB, uri) + } + case <-time.After(5 * time.Second): + t.Fatalf("client B did not receive resources/updated within 5s") + } +} + +// TestStateless_UnsubscribeTearsDown proves the other half of the +// lifecycle: Unsubscribe ends the per-URI listen stream, the server's +// UnsubscribeHandler fires, and further mutations deliver nothing. +func TestStateless_UnsubscribeTearsDown(t *testing.T) { + ctx := context.Background() + h := startHarness(t) + + notif := make(chan string, 16) + cs := h.connect(ctx, t, notif) + + task := h.createTask(ctx, t, "short-lived-watch") + uri := "tasks://" + task.Id + + if err := cs.Subscribe(ctx, &mcp.SubscribeParams{URI: uri}); err != nil { + t.Fatalf("Subscribe: %v", err) + } + waitForURI(t, h.subscribed, uri, "SubscribeHandler") + h.mutateUntilNotified(ctx, t, task, notif) + + if err := cs.Unsubscribe(ctx, &mcp.UnsubscribeParams{URI: uri}); err != nil { + t.Fatalf("Unsubscribe: %v", err) + } + waitForURI(t, h.unsubscribed, uri, "UnsubscribeHandler") + + // Drain anything in flight, then mutate and require silence. + time.Sleep(200 * time.Millisecond) + for { + select { + case <-notif: + continue + default: + } + break + } + task.Title += "!" + if _, err := h.grpcClient.UpdateTask(ctx, &tasksv1.UpdateTaskRequest{ + Id: task.Id, Title: task.Title, Description: task.Description, Done: task.Done, + }); err != nil { + t.Fatalf("UpdateTask: %v", err) + } + select { + case uri := <-notif: + t.Fatalf("received %q after Unsubscribe", uri) + case <-time.After(700 * time.Millisecond): + } +} + +// TestStateless_CancellationPropagatesToHandler proves +// PropagateRequestCancellation: aborting the HTTP request mid-call +// cancels the in-flight handler context instead of letting the handler +// run to completion for a client that is gone. +func TestStateless_CancellationPropagatesToHandler(t *testing.T) { + ctx := context.Background() + h := startHarness(t) + + started := make(chan struct{}) + canceled := make(chan struct{}) + h.srv.MustAddTool(&mcp.Tool{ + Name: "wait-for-cancel", + Description: "blocks until its context is canceled; test instrumentation", + InputSchema: protomcp.MustParseSchema(`{"type":"object"}`), + }, func(ctx context.Context, _ *mcp.CallToolRequest) (*mcp.CallToolResult, error) { + close(started) + select { + case <-ctx.Done(): + close(canceled) + return nil, ctx.Err() + case <-time.After(10 * time.Second): + return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: "never canceled"}}}, nil + } + }) + + cs := h.connect(ctx, t, nil) + callCtx, cancel := context.WithCancel(ctx) + defer cancel() + go func() { + _, _ = cs.CallTool(callCtx, &mcp.CallToolParams{ + Name: "wait-for-cancel", + Arguments: json.RawMessage(`{}`), + }) + }() + + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatalf("handler never started") + } + cancel() + select { + case <-canceled: + case <-time.After(3 * time.Second): + t.Fatalf("handler context not canceled within 3s of aborting the request") + } +} + +// TestStateless_LegacyInitializeStillServes pins backward +// compatibility: the same stateless endpoint answers a classic +// initialize from a pre-2026 client, echoing the requested legacy +// protocol version (initialize never negotiates 2026-07-28). +func TestStateless_LegacyInitializeStillServes(t *testing.T) { + h := startHarness(t) + + body := `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"legacy","version":"0.0.1"}}}` + req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, h.url, strings.NewReader(body)) + if err != nil { + t.Fatalf("new request: %v", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json, text/event-stream") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("post: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + t.Fatalf("initialize status = %d, want 200", resp.StatusCode) + } + raw, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + if !strings.Contains(string(raw), `"protocolVersion":"2025-06-18"`) { + t.Fatalf("initialize response does not echo the requested legacy protocol version: %s", raw) + } +} + +// TestStateless_RejectsGET pins the stateless transport contract: +// there is no standalone GET stream, only POSTs. +func TestStateless_RejectsGET(t *testing.T) { + h := startHarness(t) + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, h.url, nil) + if err != nil { + t.Fatalf("new request: %v", err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("get: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusMethodNotAllowed { + t.Fatalf("GET status = %d, want 405", resp.StatusCode) + } + if allow := resp.Header.Get("Allow"); allow != "POST" { + t.Fatalf("Allow = %q, want POST", allow) + } +}