fix: reassemble server-streaming google.api.HttpBody tool results instead of returning the last chunk

README.md

@@ -282,6 +282,8 @@ Multiple primitives on one RPC are legal and additive. A `GetTask` RPC annotated
 | `open_world` | optional | bool | false | Emits `annotations.openWorldHint: true`. Signals the tool reaches outside the local server (web fetch, third-party API, search) so clients can adjust consent UX. |
 | `disable_read_only_name_lint` | optional | bool | false | Suppresses the mutating-verb name lint for this RPC only. Use for false positives (`SetDifference`, `ApplyTheorem`). |
 
+**Server-streaming result semantics.** A server-streaming tool fires one `notifications/progress` event per received message and returns a summary `CallToolResult` (`N messages; last: <protojson>`) whose `structuredContent` is the last message — the right shape for watch/progress streams where each message supersedes the previous one. The exception is a stream of `google.api.HttpBody`: that is a chunked payload, so the generated handler reassembles the chunks, in order, into a single `CallToolResult`, selected by the first chunk's `content_type`: `image/*` becomes MCP image content, `audio/*` becomes audio content, and anything else must be valid UTF-8 and becomes text — a non-media payload that is not valid UTF-8 fails the call rather than returning mangled text (serve such content as a resource or out-of-band). An HttpBody-streaming tool advertises no output schema, its progress events report cumulative bytes, and `ToolResultProcessor` receives a synthesized `HttpBody` carrying the full reassembled data plus the first chunk's `content_type` and extensions. Reassembly is capped at 4 MiB by default (`protomcp.DefaultHTTPBodyDocumentLimit`) — a tool-result policy, not a transport limit: results feed a model's context window, and most hosts refuse or truncate far smaller results. A document over the cap fails the call — never silently truncates — and the cap is tunable per server via `protomcp.WithHTTPBodyDocumentLimit` (negative disables it). Note that raw byte concatenation is the HttpBody contract as served over native gRPC; grpc-gateway's streaming forwarder additionally writes a delimiter (newline by default) after each message on its HTTP responses, so servers written to compensate for that delimiter will not round-trip identically through a native-gRPC consumer like this one.
+
 ### `protomcp.v1.resource_template`, method option (read side)
 
 Emits `srv.MustAddResourceTemplate(...)`. The template is advertised via `resources/templates/list`; clients resolve a URI against it and call `resources/read`, which dispatches to your gRPC method.
@@ -450,7 +452,7 @@ Each example is standalone, runnable, and has its own README.
 
 | Example | Shows |
 |---|---|
-| [`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/greeter`](examples/greeter) | Tool primitive surface, unary + server-streaming RPCs, progress notifications with monotonic counter, **progress-token gRPC-metadata propagation**, **`google.api.HttpBody` document-stream reassembly**, `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. In-process `Hub` + `Manager` + `SubscribeHandler`/`UnsubscribeHandler` + `ss.Wait()` session-close watchdog. Race-tested. |
 | [`examples/auth`](examples/auth) | Two-layer auth: SDK-native bearer middleware **or** custom HTTP middleware, both writing gRPC metadata for the upstream |
@@ -862,7 +864,7 @@ We surveyed every Go-based proto → MCP project we could find before starting.
 - **Official SDK, exclusively.** JSON-RPC framing, session management, progress SSE, cancellation, and capability negotiation are all handled by `modelcontextprotocol/go-sdk`. Bug fixes and protocol updates flow through upstream. Redpanda ships *adapters* for both SDKs (pluggable, no default) but neither is idiomatic to their codebase; adiom uses `mark3labs/mcp-go`; linkbreakers hand-rolls the protocol with no SDK at all.
 - **Auth is a first-class extension seam, not an afterthought.** The `protomcp.ToolMiddleware` type composes like stdlib HTTP middleware but has access to both the parsed tool request **and** the outgoing gRPC metadata, so a single function can verify a caller AND propagate identity to the upstream gRPC server. None of the other three projects expose a middleware or per-request ctx hook; users have to build auth + metadata propagation on their own. Adiom exposes a `--header` CLI flag for *static* headers only.
 - **`OUTPUT_ONLY` is enforced end-to-end.** Only linkbreakers strips it from the input schema at all (via raw protowire parsing of `google.api.field_behavior`); redpanda and adiom ignore it entirely, redpanda reads `field_behavior` but only for `REQUIRED`, adiom doesn't read it at all. Nobody else runs a runtime clear. protomcp does both: strips from the schema AND runs `ClearOutputOnly` (recursive, nested, repeated, map) on every tool call so a malicious or sloppy client cannot forge server-computed fields by bypassing the advertised schema.
-- **Server-streaming is supported.** Each gRPC message becomes a `notifications/progress` event (monotonic counter per MCP spec) with a final summary `CallToolResult`. All three other projects skip streaming RPCs entirely.
+- **Server-streaming is supported.** Each gRPC message becomes a `notifications/progress` event (monotonic counter per MCP spec) with a final summary `CallToolResult`, and a stream of `google.api.HttpBody` is reassembled into one document result (text, image, or audio by content type). All three other projects skip streaming RPCs entirely.
 - **Key collisions fail loudly, at codegen and at runtime.** Both the MCP Go SDK's `AddTool` and linkbreakers' `MCPServeMux.RegisterTool` *replace* a duplicate name without warning — handler included, with no change to the advertised catalog. Our generator refuses to emit a colliding tool pair and cites both declaration sites; at runtime every registration claims its key (tool name, resource URI, URI template, prompt name) and panics on a collision, which catches what codegen cannot see: the same registrar invoked twice with different clients.
 - **Pluggable `ErrorHandler` and `ResultProcessor`.** Customize how gRPC status codes map to MCP error shapes; redact or rewrite responses after the tool handler runs. No other project offers either hook.
 - **Client-streaming / bidi annotated RPCs are hard errors.** The other three silently skip them; we surface the mistake at codegen time.

examples/greeter/e2e_test.go

@@ -9,10 +9,12 @@
 package greeter_test
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
 	"net"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -205,3 +207,241 @@ func TestServerStreamingEmitsProgress(t *testing.T) {
 	}
 	_ = out // silence the out-unused warning in the non-race branch
 }
+
+func TestHTTPBodyStreamToolReassemblesTheDocument(t *testing.T) {
+	grpcClient := startGRPC(t)
+	srv := protomcp.New("greeter", "0.1.0")
+	greeterv1.RegisterGreeterMCPTools(srv, grpcClient)
+
+	ctx := context.Background()
+	cs := connect(ctx, t, srv, nil)
+
+	turns := 5
+	out, err := cs.CallTool(ctx, &mcp.CallToolParams{
+		Name:      "Greeter_DownloadTranscript",
+		Arguments: json.RawMessage(fmt.Sprintf(`{"name":"world","turns":%d}`, turns)),
+	})
+	if err != nil {
+		t.Fatalf("call: %v", err)
+	}
+	if out.IsError {
+		t.Fatalf("unexpected IsError: %+v", out)
+	}
+
+	want := ""
+	for i := 1; i <= turns; i++ {
+		want += fmt.Sprintf("Turn %d: hello, world!\n", i)
+	}
+	if len(out.Content) != 1 {
+		t.Fatalf("content items: got %d, want 1 (%+v)", len(out.Content), out.Content)
+	}
+	text, ok := out.Content[0].(*mcp.TextContent)
+	if !ok {
+		t.Fatalf("content type: got %T, want *mcp.TextContent", out.Content[0])
+	}
+	if text.Text != want {
+		t.Errorf("document: got %q, want %q", text.Text, want)
+	}
+	if out.StructuredContent != nil {
+		t.Errorf("document tool must not return structured content, got %v", out.StructuredContent)
+	}
+}
+
+func TestHTTPBodyStreamToolAdvertisesNoOutputSchema(t *testing.T) {
+	grpcClient := startGRPC(t)
+	srv := protomcp.New("greeter", "0.1.0")
+	greeterv1.RegisterGreeterMCPTools(srv, grpcClient)
+
+	ctx := context.Background()
+	cs := connect(ctx, t, srv, nil)
+
+	list, err := cs.ListTools(ctx, nil)
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	for _, tt := range list.Tools {
+		if tt.Name != "Greeter_DownloadTranscript" {
+			continue
+		}
+		if tt.OutputSchema != nil {
+			t.Errorf("document tool must advertise no output schema, got %v", tt.OutputSchema)
+		}
+		if tt.InputSchema == nil {
+			t.Errorf("document tool must still advertise its input schema")
+		}
+		return
+	}
+	t.Fatalf("Greeter_DownloadTranscript missing from tools/list")
+}
+
+func TestHTTPBodyStreamToolRejectsNonUTF8(t *testing.T) {
+	grpcClient := startGRPC(t)
+	srv := protomcp.New("greeter", "0.1.0")
+	greeterv1.RegisterGreeterMCPTools(srv, grpcClient)
+
+	ctx := context.Background()
+	cs := connect(ctx, t, srv, nil)
+
+	out, err := cs.CallTool(ctx, &mcp.CallToolParams{
+		Name:      "Greeter_DownloadTranscript",
+		Arguments: json.RawMessage(`{"name":"world","binary":true}`),
+	})
+	if err != nil {
+		t.Fatalf("call: %v", err)
+	}
+	if !out.IsError {
+		t.Fatalf("a non-UTF-8 payload must produce a tool error, got %+v", out)
+	}
+	text, ok := out.Content[0].(*mcp.TextContent)
+	if !ok {
+		t.Fatalf("content type: got %T, want *mcp.TextContent", out.Content[0])
+	}
+	if !strings.Contains(text.Text, "valid UTF-8") {
+		t.Errorf("error text should name the UTF-8 policy, got %q", text.Text)
+	}
+}
+
+func TestHTTPBodyStreamToolMapsMediaContentTypes(t *testing.T) {
+	grpcClient := startGRPC(t)
+	srv := protomcp.New("greeter", "0.1.0")
+	greeterv1.RegisterGreeterMCPTools(srv, grpcClient)
+
+	ctx := context.Background()
+	cs := connect(ctx, t, srv, nil)
+
+	out, err := cs.CallTool(ctx, &mcp.CallToolParams{
+		Name:      "Greeter_DownloadTranscript",
+		Arguments: json.RawMessage(`{"name":"world","binary":true,"contentType":"image/png"}`),
+	})
+	if err != nil {
+		t.Fatalf("call: %v", err)
+	}
+	if out.IsError {
+		t.Fatalf("a media payload must not be rejected for being non-UTF-8, got %+v", out)
+	}
+	img, ok := out.Content[0].(*mcp.ImageContent)
+	if !ok {
+		t.Fatalf("content type: got %T, want *mcp.ImageContent", out.Content[0])
+	}
+	if img.MIMEType != "image/png" {
+		t.Errorf("MIMEType: got %q, want %q", img.MIMEType, "image/png")
+	}
+	if !bytes.Equal(img.Data, []byte{0xff, 0xfe, 0x00, 0x01}) {
+		t.Errorf("image data: got %v, want the server's raw bytes", img.Data)
+	}
+}
+
+func TestHTTPBodyStreamToolReportsCumulativeByteProgress(t *testing.T) {
+	grpcClient := startGRPC(t)
+	srv := protomcp.New("greeter", "0.1.0")
+	greeterv1.RegisterGreeterMCPTools(srv, grpcClient)
+
+	var (
+		mu       sync.Mutex
+		progress []string
+		counters []float64
+	)
+	opts := &mcp.ClientOptions{
+		ProgressNotificationHandler: func(_ context.Context, p *mcp.ProgressNotificationClientRequest) {
+			mu.Lock()
+			defer mu.Unlock()
+			progress = append(progress, p.Params.Message)
+			counters = append(counters, p.Params.Progress)
+		},
+	}
+
+	ctx := context.Background()
+	cs := connect(ctx, t, srv, opts)
+
+	turns := 3
+	transcript := ""
+	for i := 1; i <= turns; i++ {
+		transcript += fmt.Sprintf("Turn %d: hello, %s!\n", i, "world")
+	}
+	const chunkSize = 10
+	wantChunks := (len(transcript) + chunkSize - 1) / chunkSize
+
+	out, err := cs.CallTool(ctx, &mcp.CallToolParams{
+		Name:      "Greeter_DownloadTranscript",
+		Arguments: json.RawMessage(fmt.Sprintf(`{"name":"world","turns":%d}`, turns)),
+		Meta:      mcp.Meta{"progressToken": "transcript-progress-1"},
+	})
+	if err != nil {
+		t.Fatalf("call: %v", err)
+	}
+	if out.IsError {
+		t.Fatalf("unexpected IsError: %+v", out)
+	}
+
+	var got []string
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		mu.Lock()
+		got = append([]string(nil), progress...)
+		mu.Unlock()
+		if len(got) >= wantChunks {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if len(got) != wantChunks {
+		t.Fatalf("progress events: got %d, want %d (msgs=%v)", len(got), wantChunks, got)
+	}
+	if last := got[len(got)-1]; last != fmt.Sprintf("%d bytes", len(transcript)) {
+		t.Errorf("final progress message: got %q, want %q", last, fmt.Sprintf("%d bytes", len(transcript)))
+	}
+	for i, m := range got {
+		if !strings.HasSuffix(m, " bytes") {
+			t.Errorf("progress[%d] should report cumulative bytes, got %q", i, m)
+		}
+	}
+
+	mu.Lock()
+	gotCounters := append([]float64(nil), counters...)
+	mu.Unlock()
+	for i := 1; i < len(gotCounters); i++ {
+		if gotCounters[i] <= gotCounters[i-1] {
+			t.Errorf("progress counter not monotonic: [%d]=%v [%d]=%v (full=%v)",
+				i-1, gotCounters[i-1], i, gotCounters[i], gotCounters)
+			break
+		}
+	}
+}
+
+func TestHTTPBodyStreamToolEnforcesTheDocumentLimit(t *testing.T) {
+	grpcClient := startGRPC(t)
+	srv := protomcp.New("greeter", "0.1.0", protomcp.WithHTTPBodyDocumentLimit(32))
+	greeterv1.RegisterGreeterMCPTools(srv, grpcClient)
+
+	ctx := context.Background()
+	cs := connect(ctx, t, srv, nil)
+
+	out, err := cs.CallTool(ctx, &mcp.CallToolParams{
+		Name:      "Greeter_DownloadTranscript",
+		Arguments: json.RawMessage(`{"name":"world","turns":5}`),
+	})
+	if err != nil {
+		t.Fatalf("call: %v", err)
+	}
+	if !out.IsError {
+		t.Fatalf("a document above the limit must produce a tool error, got %+v", out)
+	}
+	text, ok := out.Content[0].(*mcp.TextContent)
+	if !ok {
+		t.Fatalf("content type: got %T, want *mcp.TextContent", out.Content[0])
+	}
+	if !strings.Contains(text.Text, "32-byte limit") {
+		t.Errorf("error text should name the limit, got %q", text.Text)
+	}
+
+	within, err := cs.CallTool(ctx, &mcp.CallToolParams{
+		Name:      "Greeter_DownloadTranscript",
+		Arguments: json.RawMessage(`{"name":"jo","turns":1}`),
+	})
+	if err != nil {
+		t.Fatalf("call: %v", err)
+	}
+	if within.IsError {
+		t.Fatalf("a document within the limit must succeed, got %+v", within)
+	}
+}

examples/greeter/server/server.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"fmt"
 
+	"google.golang.org/genproto/googleapis/api/httpbody"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -92,6 +93,38 @@ func (s *Server) Slow(ctx context.Context, _ *greeterv1.HelloRequest) (*greeterv
 	return nil, status.Error(codes.Canceled, "canceled")
 }
 
+func (s *Server) DownloadTranscript(req *greeterv1.DownloadTranscriptRequest, stream greeterv1.Greeter_DownloadTranscriptServer) error {
+	if req.GetBinary() {
+		contentType := req.GetContentType()
+		if contentType == "" {
+			contentType = "application/octet-stream"
+		}
+		return stream.Send(&httpbody.HttpBody{
+			ContentType: contentType,
+			Data:        []byte{0xff, 0xfe, 0x00, 0x01},
+		})
+	}
+	turns := int(req.GetTurns())
+	if turns <= 0 {
+		turns = 1
+	}
+	var transcript string
+	for i := 1; i <= turns; i++ {
+		transcript += fmt.Sprintf("Turn %d: hello, %s!\n", i, req.GetName())
+	}
+	const chunkSize = 10
+	for start := 0; start < len(transcript); start += chunkSize {
+		end := min(start+chunkSize, len(transcript))
+		if err := stream.Send(&httpbody.HttpBody{
+			ContentType: "text/plain; charset=utf-8",
+			Data:        []byte(transcript[start:end]),
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (s *Server) Internal(_ context.Context, req *greeterv1.HelloRequest) (*greeterv1.HelloReply, error) {
 	return &greeterv1.HelloReply{Message: "internal-only: " + req.GetName()}, nil
 }