Repository: modelcontextprotocol/go-sdk

Issue:
Streamable HTTP support
This issue tracks work on the Streamable HTTP transport:

- [x] Add a `StreamableServerTransport` type.
- [x] Add a `StreamableClientTransport` type.
- [x] Support 'stateless' mode for server transport (not documented, but implicit: see https://github.com/modelcontextprotocol/modelcontextprotocol/issues/282#issuecomment-2998128187).
- [x] Add configurable event store for the streamable server transport (see TypeScript's 'EventStore').
- [x] Support retryability/replayability for the streamable client transport.
- [x] Support a single JSON response (application/json)

Discussion:
@samthanawalla: I can take a look at "Support retryability/replayability for the streamable client transport."@findleyr: @samthanawalla sure, that would be great. I started thinking about persistence in https://go.dev/cl/682555, but haven't done anything toward retryability. I think there is some good prior art in other SDKs. For example:
https://github.com/modelcontextprotocol/typescript-sdk/blob/main/src/client/streamableHttp.ts#L51

I will work on server persistence APIs.@samthanawalla: We also need to support application/json #129 @jpmcb: > Support 'stateless' mode for server transport

I'd be happy to take a look at this since it's important for alot of servers I've been building.

Adding an option to `ServerOptions` to not expect session management seems like an ideal solution. This is how alot of other SDKs implement this (like in the Python SDK):

```python
from mcp.server.fastmcp import FastMCP

# Stateless server (no session persistence)
mcp = FastMCP("StatelessServer", stateless_http=True)
```@findleyr: @jpmcb thank you, and sorry for the missing support. I've been away for a few weeks, but @samthanawalla, @jba and I have a meeting tomorrow to talk about outstanding items, of which streamable transport is the most important.  I think it's likely that we'll prioritize this above essentially everything else, and with Sam working on configurable session storage, it makes sense for us to do stateless mode as well.@whiletrue111: So does it support streamable HTTP?
@jba: > So does it support streamable HTTP?

Yes, at main.@findleyr: Stateless support is next after #257.@findleyr: This will be also be completed by #307.

Solution patch:
diff --git a/jsonschema/infer.go b/jsonschema/infer.go
index 654e619..9ff0ddd 100644
--- a/jsonschema/infer.go
+++ b/jsonschema/infer.go
@@ -47,7 +47,7 @@ import (
 func For[T any]() (*Schema, error) {
 	// TODO: consider skipping incompatible fields, instead of failing.
 	seen := make(map[reflect.Type]bool)
-	s, err := forType(reflect.TypeFor[T](), seen)
+	s, err := forType(reflect.TypeFor[T](), seen, false)
 	if err != nil {
 		var z T
 		return nil, fmt.Errorf("For[%T](): %w", z, err)
@@ -55,7 +55,22 @@ func For[T any]() (*Schema, error) {
 	return s, nil
 }
 
-func forType(t reflect.Type, seen map[reflect.Type]bool) (*Schema, error) {
+// ForLax behaves like [For], except that it ignores struct fields with invalid types instead of
+// returning an error. That allows callers to adjust the resulting schema using custom knowledge.
+// For example, an interface type where all the possible implementations are known
+// can be described with "oneof".
+func ForLax[T any]() (*Schema, error) {
+	// TODO: consider skipping incompatible fields, instead of failing.
+	seen := make(map[reflect.Type]bool)
+	s, err := forType(reflect.TypeFor[T](), seen, true)
+	if err != nil {
+		var z T
+		return nil, fmt.Errorf("ForLax[%T](): %w", z, err)
+	}
+	return s, nil
+}
+
+func forType(t reflect.Type, seen map[reflect.Type]bool, lax bool) (*Schema, error) {
 	// Follow pointers: the schema for *T is almost the same as for T, except that
 	// an explicit JSON "null" is allowed for the pointer.
 	allowNull := false
@@ -96,20 +111,33 @@ func forType(t reflect.Type, seen map[reflect.Type]bool) (*Schema, error) {
 
 	case reflect.Map:
 		if t.Key().Kind() != reflect.String {
+			if lax {
+				return nil, nil // ignore
+			}
 			return nil, fmt.Errorf("unsupported map key type %v", t.Key().Kind())
 		}
+		if t.Key().Kind() != reflect.String {
+		}
 		s.Type = "object"
-		s.AdditionalProperties, err = forType(t.Elem(), seen)
+		s.AdditionalProperties, err = forType(t.Elem(), seen, lax)
 		if err != nil {
 			return nil, fmt.Errorf("computing map value schema: %v", err)
 		}
+		if lax && s.AdditionalProperties == nil {
+			// Ignore if the element type is invalid.
+			return nil, nil
+		}
 
 	case reflect.Slice, reflect.Array:
 		s.Type = "array"
-		s.Items, err = forType(t.Elem(), seen)
+		s.Items, err = forType(t.Elem(), seen, lax)
 		if err != nil {
 			return nil, fmt.Errorf("computing element schema: %v", err)
 		}
+		if lax && s.Items == nil {
+			// Ignore if the element type is invalid.
+			return nil, nil
+		}
 		if t.Kind() == reflect.Array {
 			s.MinItems = Ptr(t.Len())
 			s.MaxItems = Ptr(t.Len())
@@ -132,10 +160,14 @@ func forType(t reflect.Type, seen map[reflect.Type]bool) (*Schema, error) {
 			if s.Properties == nil {
 				s.Properties = make(map[string]*Schema)
 			}
-			fs, err := forType(field.Type, seen)
+			fs, err := forType(field.Type, seen, lax)
 			if err != nil {
 				return nil, err
 			}
+			if lax && fs == nil {
+				// Skip fields of invalid type.
+				continue
+			}
 			if tag, ok := field.Tag.Lookup("jsonschema"); ok {
 				if tag == "" {
 					return nil, fmt.Errorf("empty jsonschema tag on struct field %s.%s", t, field.Name)
@@ -152,6 +184,10 @@ func forType(t reflect.Type, seen map[reflect.Type]bool) (*Schema, error) {
 		}
 
 	default:
+		if lax {
+			// Ignore.
+			return nil, nil
+		}
 		return nil, fmt.Errorf("type %v is unsupported by jsonschema", t)
 	}
 	if allowNull && s.Type != "" {
diff --git a/mcp/content.go b/mcp/content.go
index 04888de..8bf75f0 100644
--- a/mcp/content.go
+++ b/mcp/content.go
@@ -63,13 +63,7 @@ func (c *ImageContent) MarshalJSON() ([]byte, error) {
 	if data == nil {
 		data = []byte{}
 	}
-	wire := struct {
-		Type        string       `json:"type"`
-		MIMEType    string       `json:"mimeType"`
-		Data        []byte       `json:"data"`
-		Meta        Meta         `json:"_meta,omitempty"`
-		Annotations *Annotations `json:"annotations,omitempty"`
-	}{
+	wire := imageAudioWire{
 		Type:        "image",
 		MIMEType:    c.MIMEType,
 		Data:        data,
@@ -100,13 +94,7 @@ func (c AudioContent) MarshalJSON() ([]byte, error) {
 	if data == nil {
 		data = []byte{}
 	}
-	wire := struct {
-		Type        string       `json:"type"`
-		MIMEType    string       `json:"mimeType"`
-		Data        []byte       `json:"data"`
-		Meta        Meta         `json:"_meta,omitempty"`
-		Annotations *Annotations `json:"annotations,omitempty"`
-	}{
+	wire := imageAudioWire{
 		Type:        "audio",
 		MIMEType:    c.MIMEType,
 		Data:        data,
@@ -123,6 +111,15 @@ func (c *AudioContent) fromWire(wire *wireContent) {
 	c.Annotations = wire.Annotations
 }
 
+// Custom wire format to ensure required fields are always included, even when empty.
+type imageAudioWire struct {
+	Type        string       `json:"type"`
+	MIMEType    string       `json:"mimeType"`
+	Data        []byte       `json:"data"`
+	Meta        Meta         `json:"_meta,omitempty"`
+	Annotations *Annotations `json:"annotations,omitempty"`
+}
+
 // ResourceLink is a link to a resource
 type ResourceLink struct {
 	URI         string
diff --git a/mcp/event.go b/mcp/event.go
index 4562b81..fbbe194 100644
--- a/mcp/event.go
+++ b/mcp/event.go
@@ -10,14 +10,22 @@ package mcp
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
 	"iter"
+	"maps"
 	"net/http"
+	"slices"
 	"strings"
+	"sync"
 )
 
+// If true, MemoryEventStore will do frequent validation to check invariants, slowing it down.
+// Remove when we're confident in the code.
+const validateMemoryEventStore = true
+
 // An Event is a server-sent event.
 // See https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#fields.
 type Event struct {
@@ -136,3 +144,285 @@ func scanEvents(r io.Reader) iter.Seq2[Event, error] {
 		}
 	}
 }
+
+// An EventStore tracks data for SSE streams.
+// A single EventStore suffices for all sessions, since session IDs are
+// globally unique. So one EventStore can be created per process, for
+// all Servers in the process.
+// Such a store is able to bound resource usage for the entire process.
+//
+// All of an EventStore's methods must be safe for use by multiple goroutines.
+type EventStore interface {
+	// AppendEvent appends data for an outgoing event to given stream, which is part of the
+	// given session. It returns the index of the event in the stream, suitable for constructing
+	// an event ID to send to the client.
+	AppendEvent(_ context.Context, sessionID string, _ StreamID, data []byte) (int, error)
+
+	// After returns an iterator over the data for the given session and stream, beginning
+	// just after the given index.
+	// Once the iterator yields a non-nil error, it will stop.
+	// After's iterator must return an error immediately if any data after index was
+	// dropped; it must not return partial results.
+	After(_ context.Context, sessionID string, _ StreamID, index int) iter.Seq2[[]byte, error]
+
+	// StreamClosed informs the store that the given stream is finished.
+	// A store cannot rely on this method being called for cleanup. It should institute
+	// additional mechanisms, such as timeouts, to reclaim storage.
+	StreamClosed(_ context.Context, sessionID string, streamID StreamID) error
+
+	// SessionClosed informs the store that the given session is finished, along
+	// with all of its streams.
+	// A store cannot rely on this method being called for cleanup. It should institute
+	// additional mechanisms, such as timeouts, to reclaim storage.
+	SessionClosed(_ context.Context, sessionID string) error
+}
+
+// A dataList is a list of []byte.
+// The zero dataList is ready to use.
+type dataList struct {
+	size  int // total size of data bytes
+	first int // the stream index of the first element in data
+	data  [][]byte
+}
+
+func (dl *dataList) appendData(d []byte) {
+	// If we allowed empty data, we would consume memory without incrementing the size.
+	// We could of course account for that, but we keep it simple and assume there is no
+	// empty data.
+	if len(d) == 0 {
+		panic("empty data item")
+	}
+	dl.data = append(dl.data, d)
+	dl.size += len(d)
+}
+
+// removeFirst removes the first data item in dl, returning the size of the item.
+// It panics if dl is empty.
+func (dl *dataList) removeFirst() int {
+	if len(dl.data) == 0 {
+		panic("empty dataList")
+	}
+	r := len(dl.data[0])
+	dl.size -= r
+	dl.data[0] = nil // help GC
+	dl.data = dl.data[1:]
+	dl.first++
+	return r
+}
+
+// lastIndex returns the index of the last data item in dl.
+// It panics if there are none.
+func (dl *dataList) lastIndex() int {
+	if len(dl.data) == 0 {
+		panic("empty dataList")
+	}
+	return dl.first + len(dl.data) - 1
+}
+
+// A MemoryEventStore is an [EventStore] backed by memory.
+type MemoryEventStore struct {
+	mu       sync.Mutex
+	maxBytes int                               // max total size of all data
+	nBytes   int                               // current total size of all data
+	store    map[string]map[StreamID]*dataList // session ID -> stream ID -> *dataList
+}
+
+// MemoryEventStoreOptions are options for a [MemoryEventStore].
+type MemoryEventStoreOptions struct{}
+
+// MaxBytes returns the maximum number of bytes that the store will retain before
+// purging data.
+func (s *MemoryEventStore) MaxBytes() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.maxBytes
+}
+
+// SetMaxBytes sets the maximum number of bytes the store will retain before purging
+// data. The argument must not be negative. If it is zero, a suitable default will be used.
+// SetMaxBytes can be called at any time. The size of the store will be adjusted
+// immediately.
+func (s *MemoryEventStore) SetMaxBytes(n int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	switch {
+	case n < 0:
+		panic("negative argument")
+	case n == 0:
+		s.maxBytes = defaultMaxBytes
+	default:
+		s.maxBytes = n
+	}
+	s.purge()
+}
+
+const defaultMaxBytes = 10 << 20 // 10 MiB
+
+// NewMemoryEventStore creates a [MemoryEventStore] with the default value
+// for MaxBytes.
+func NewMemoryEventStore(opts *MemoryEventStoreOptions) *MemoryEventStore {
+	return &MemoryEventStore{
+		maxBytes: defaultMaxBytes,
+		store:    make(map[string]map[StreamID]*dataList),
+	}
+}
+
+// AppendEvent implements [EventStore.AppendEvent] by recording data
+// in memory.
+func (s *MemoryEventStore) AppendEvent(_ context.Context, sessionID string, streamID StreamID, data []byte) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	streamMap, ok := s.store[sessionID]
+	if !ok {
+		streamMap = make(map[StreamID]*dataList)
+		s.store[sessionID] = streamMap
+	}
+	dl, ok := streamMap[streamID]
+	if !ok {
+		dl = &dataList{}
+		streamMap[streamID] = dl
+	}
+	// Purge before adding, so at least the current data item will be present.
+	// (That could result in nBytes > maxBytes, but we'll live with that.)
+	s.purge()
+	dl.appendData(data)
+	s.nBytes += len(data)
+	return dl.lastIndex(), nil
+}
+
+// After implements [EventStore.After].
+func (s *MemoryEventStore) After(_ context.Context, sessionID string, streamID StreamID, index int) iter.Seq2[[]byte, error] {
+	// Return the data items to yield.
+	// We must copy, because dataList.removeFirst nils out slice elements.
+	copyData := func() ([][]byte, error) {
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		streamMap, ok := s.store[sessionID]
+		if !ok {
+			return nil, fmt.Errorf("MemoryEventStore.After: unknown session ID %q", sessionID)
+		}
+		dl, ok := streamMap[streamID]
+		if !ok {
+			return nil, fmt.Errorf("MemoryEventStore.After: unknown stream ID %v in session %q", streamID, sessionID)
+		}
+		if dl.first > index {
+			return nil, fmt.Errorf("MemoryEventStore.After: data purged at index %d, stream ID %v, session %q", index, streamID, sessionID)
+		}
+		return slices.Clone(dl.data[index-dl.first:]), nil
+	}
+
+	return func(yield func([]byte, error) bool) {
+		ds, err := copyData()
+		if err != nil {
+			yield(nil, err)
+			return
+		}
+		for _, d := range ds {
+			if !yield(d, nil) {
+				return
+			}
+		}
+	}
+}
+
+// StreamClosed implements [EventStore.StreamClosed].
+func (s *MemoryEventStore) StreamClosed(_ context.Context, sessionID string, streamID StreamID) error {
+	if sessionID == "" {
+		panic("empty sessionID")
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	sm := s.store[sessionID]
+	dl := sm[streamID]
+	s.nBytes -= dl.size
+	delete(sm, streamID)
+	if len(sm) == 0 {
+		delete(s.store, sessionID)
+	}
+	s.validate()
+	return nil
+}
+
+// SessionClosed implements [EventStore.SessionClosed].
+func (s *MemoryEventStore) SessionClosed(_ context.Context, sessionID string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, dl := range s.store[sessionID] {
+		s.nBytes -= dl.size
+	}
+	delete(s.store, sessionID)
+	s.validate()
+	return nil
+}
+
+// purge removes data until no more than s.maxBytes bytes are in use.
+// It must be called with s.mu held.
+func (s *MemoryEventStore) purge() {
+	// Remove the first element of every dataList until below the max.
+	for s.nBytes > s.maxBytes {
+		changed := false
+		for _, sm := range s.store {
+			for _, dl := range sm {
+				if dl.size > 0 {
+					r := dl.removeFirst()
+					if r > 0 {
+						changed = true
+						s.nBytes -= r
+					}
+				}
+			}
+		}
+		if !changed {
+			panic("no progress during purge")
+		}
+	}
+	s.validate()
+}
+
+// validate checks that the store's data structures are valid.
+// It must be called with s.mu held.
+func (s *MemoryEventStore) validate() {
+	if !validateMemoryEventStore {
+		return
+	}
+	// Check that we're accounting for the size correctly.
+	n := 0
+	for _, sm := range s.store {
+		for _, dl := range sm {
+			for _, d := range dl.data {
+				n += len(d)
+			}
+		}
+	}
+	if n != s.nBytes {
+		panic("sizes don't add up")
+	}
+}
+
+// debugString returns a string containing the state of s.
+// Used in tests.
+func (s *MemoryEventStore) debugString() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var b strings.Builder
+	for i, sess := range slices.Sorted(maps.Keys(s.store)) {
+		if i > 0 {
+			fmt.Fprintf(&b, "; ")
+		}
+		sm := s.store[sess]
+		for i, sid := range slices.Sorted(maps.Keys(sm)) {
+			if i > 0 {
+				fmt.Fprintf(&b, "; ")
+			}
+			dl := sm[sid]
+			fmt.Fprintf(&b, "%s %d first=%d", sess, sid, dl.first)
+			for _, d := range dl.data {
+				fmt.Fprintf(&b, " %s", d)
+			}
+		}
+	}
+	return b.String()
+}
diff --git a/mcp/streamable.go b/mcp/streamable.go
index ef740d3..728da85 100644
--- a/mcp/streamable.go
+++ b/mcp/streamable.go
@@ -163,10 +163,10 @@ func NewStreamableServerTransport(sessionID string) *StreamableServerTransport {
 		id:               sessionID,
 		incoming:         make(chan jsonrpc.Message, 10),
 		done:             make(chan struct{}),
-		outgoingMessages: make(map[streamID][]*streamableMsg),
-		signals:          make(map[streamID]chan struct{}),
-		requestStreams:   make(map[jsonrpc.ID]streamID),
-		streamRequests:   make(map[streamID]map[jsonrpc.ID]struct{}),
+		outgoingMessages: make(map[StreamID][]*streamableMsg),
+		signals:          make(map[StreamID]chan struct{}),
+		requestStreams:   make(map[jsonrpc.ID]StreamID),
+		streamRequests:   make(map[StreamID]map[jsonrpc.ID]struct{}),
 	}
 }
 
@@ -212,7 +212,7 @@ type StreamableServerTransport struct {
 	//
 	// TODO(rfindley): garbage collect this data. For now, we save all outgoingMessages
 	// messages for the lifespan of the transport.
-	outgoingMessages map[streamID][]*streamableMsg
+	outgoingMessages map[StreamID][]*streamableMsg
 
 	// signals maps a logical stream ID to a 1-buffered channel, owned by an
 	// incoming HTTP request, that signals that there are messages available to
@@ -223,14 +223,14 @@ type StreamableServerTransport struct {
 	//
 	// Lifecycle: signals persists for the duration of an HTTP POST or GET
 	// request for the given streamID.
-	signals map[streamID]chan struct{}
+	signals map[StreamID]chan struct{}
 
 	// requestStreams maps incoming requests to their logical stream ID.
 	//
 	// Lifecycle: requestStreams persists for the duration of the session.
 	//
 	// TODO(rfindley): clean up once requests are handled.
-	requestStreams map[jsonrpc.ID]streamID
+	requestStreams map[jsonrpc.ID]StreamID
 
 	// streamRequests tracks the set of unanswered incoming RPCs for each logical
 	// stream.
@@ -241,10 +241,10 @@ type StreamableServerTransport struct {
 	// Lifecycle: streamRequests values persist as until the requests have been
 	// replied to by the server. Notably, NOT until they are sent to an HTTP
 	// response, as delivery is not guaranteed.
-	streamRequests map[streamID]map[jsonrpc.ID]struct{}
+	streamRequests map[StreamID]map[jsonrpc.ID]struct{}
 }
 
-type streamID int64
+type StreamID int64
 
 // a streamableMsg is an SSE event with an index into its logical stream.
 type streamableMsg struct {
@@ -298,7 +298,7 @@ func (t *StreamableServerTransport) ServeHTTP(w http.ResponseWriter, req *http.R
 
 func (t *StreamableServerTransport) serveGET(w http.ResponseWriter, req *http.Request) {
 	// connID 0 corresponds to the default GET request.
-	id, nextIdx := streamID(0), 0
+	id, nextIdx := StreamID(0), 0
 	if len(req.Header.Values("Last-Event-ID")) > 0 {
 		eid := req.Header.Get("Last-Event-ID")
 		var ok bool
@@ -352,7 +352,7 @@ func (t *StreamableServerTransport) servePOST(w http.ResponseWriter, req *http.R
 	}
 
 	// Update accounting for this request.
-	id := streamID(t.nextStreamID.Add(1))
+	id := StreamID(t.nextStreamID.Add(1))
 	signal := make(chan struct{}, 1)
 	t.mu.Lock()
 	if len(requests) > 0 {
@@ -376,7 +376,7 @@ func (t *StreamableServerTransport) servePOST(w http.ResponseWriter, req *http.R
 	t.streamResponse(w, req, id, 0, signal)
 }
 
-func (t *StreamableServerTransport) streamResponse(w http.ResponseWriter, req *http.Request, id streamID, nextIndex int, signal chan struct{}) {
+func (t *StreamableServerTransport) streamResponse(w http.ResponseWriter, req *http.Request, id StreamID, nextIndex int, signal chan struct{}) {
 	defer func() {
 		t.mu.Lock()
 		delete(t.signals, id)
@@ -463,7 +463,7 @@ stream:
 // streamID and message index idx.
 //
 // See also [parseEventID].
-func formatEventID(sid streamID, idx int) string {
+func formatEventID(sid StreamID, idx int) string {
 	return fmt.Sprintf("%d_%d", sid, idx)
 }
 
@@ -471,7 +471,7 @@ func formatEventID(sid streamID, idx int) string {
 // index.
 //
 // See also [formatEventID].
-func parseEventID(eventID string) (sid streamID, idx int, ok bool) {
+func parseEventID(eventID string) (sid StreamID, idx int, ok bool) {
 	parts := strings.Split(eventID, "_")
 	if len(parts) != 2 {
 		return 0, 0, false
@@ -484,7 +484,7 @@ func parseEventID(eventID string) (sid streamID, idx int, ok bool) {
 	if err != nil || idx < 0 {
 		return 0, 0, false
 	}
-	return streamID(stream), idx, true
+	return StreamID(stream), idx, true
 }
 
 // Read implements the [Connection] interface.
@@ -523,7 +523,7 @@ func (t *StreamableServerTransport) Write(ctx context.Context, msg jsonrpc.Messa
 	//
 	// For messages sent outside of a request context, this is the default
 	// connection 0.
-	var forConn streamID
+	var forConn StreamID
 	if forRequest.IsValid() {
 		t.mu.Lock()
 		forConn = t.requestStreams[forRequest]



Test patch:
diff --git a/jsonschema/infer_test.go b/jsonschema/infer_test.go
index 106e537..8c8feec 100644
--- a/jsonschema/infer_test.go
+++ b/jsonschema/infer_test.go
@@ -13,8 +13,14 @@ import (
 	"github.com/modelcontextprotocol/go-sdk/jsonschema"
 )
 
-func forType[T any]() *jsonschema.Schema {
-	s, err := jsonschema.For[T]()
+func forType[T any](lax bool) *jsonschema.Schema {
+	var s *jsonschema.Schema
+	var err error
+	if lax {
+		s, err = jsonschema.ForLax[T]()
+	} else {
+		s, err = jsonschema.For[T]()
+	}
 	if err != nil {
 		panic(err)
 	}
@@ -28,104 +34,134 @@ func TestFor(t *testing.T) {
 		B int `jsonschema:"bdesc"`
 	}
 
-	tests := []struct {
+	type test struct {
 		name string
 		got  *jsonschema.Schema
 		want *jsonschema.Schema
-	}{
-		{"string", forType[string](), &schema{Type: "string"}},
-		{"int", forType[int](), &schema{Type: "integer"}},
-		{"int16", forType[int16](), &schema{Type: "integer"}},
-		{"uint32", forType[int16](), &schema{Type: "integer"}},
-		{"float64", forType[float64](), &schema{Type: "number"}},
-		{"bool", forType[bool](), &schema{Type: "boolean"}},
-		{"intmap", forType[map[string]int](), &schema{
-			Type:                 "object",
-			AdditionalProperties: &schema{Type: "integer"},
-		}},
-		{"anymap", forType[map[string]any](), &schema{
-			Type:                 "object",
-			AdditionalProperties: &schema{},
-		}},
-		{
-			"struct",
-			forType[struct {
-				F           int `json:"f" jsonschema:"fdesc"`
-				G           []float64
-				P           *bool  `jsonschema:"pdesc"`
-				Skip        string `json:"-"`
-				NoSkip      string `json:",omitempty"`
-				unexported  float64
-				unexported2 int `json:"No"`
-			}](),
-			&schema{
-				Type: "object",
-				Properties: map[string]*schema{
-					"f":      {Type: "integer", Description: "fdesc"},
-					"G":      {Type: "array", Items: &schema{Type: "number"}},
-					"P":      {Types: []string{"null", "boolean"}, Description: "pdesc"},
-					"NoSkip": {Type: "string"},
+	}
+
+	tests := func(lax bool) []test {
+		return []test{
+			{"string", forType[string](lax), &schema{Type: "string"}},
+			{"int", forType[int](lax), &schema{Type: "integer"}},
+			{"int16", forType[int16](lax), &schema{Type: "integer"}},
+			{"uint32", forType[int16](lax), &schema{Type: "integer"}},
+			{"float64", forType[float64](lax), &schema{Type: "number"}},
+			{"bool", forType[bool](lax), &schema{Type: "boolean"}},
+			{"intmap", forType[map[string]int](lax), &schema{
+				Type:                 "object",
+				AdditionalProperties: &schema{Type: "integer"},
+			}},
+			{"anymap", forType[map[string]any](lax), &schema{
+				Type:                 "object",
+				AdditionalProperties: &schema{},
+			}},
+			{
+				"struct",
+				forType[struct {
+					F           int `json:"f" jsonschema:"fdesc"`
+					G           []float64
+					P           *bool  `jsonschema:"pdesc"`
+					Skip        string `json:"-"`
+					NoSkip      string `json:",omitempty"`
+					unexported  float64
+					unexported2 int `json:"No"`
+				}](lax),
+				&schema{
+					Type: "object",
+					Properties: map[string]*schema{
+						"f":      {Type: "integer", Description: "fdesc"},
+						"G":      {Type: "array", Items: &schema{Type: "number"}},
+						"P":      {Types: []string{"null", "boolean"}, Description: "pdesc"},
+						"NoSkip": {Type: "string"},
+					},
+					Required:             []string{"f", "G", "P"},
+					AdditionalProperties: falseSchema(),
 				},
-				Required:             []string{"f", "G", "P"},
-				AdditionalProperties: falseSchema(),
 			},
-		},
-		{
-			"no sharing",
-			forType[struct{ X, Y int }](),
-			&schema{
-				Type: "object",
-				Properties: map[string]*schema{
-					"X": {Type: "integer"},
-					"Y": {Type: "integer"},
+			{
+				"no sharing",
+				forType[struct{ X, Y int }](lax),
+				&schema{
+					Type: "object",
+					Properties: map[string]*schema{
+						"X": {Type: "integer"},
+						"Y": {Type: "integer"},
+					},
+					Required:             []string{"X", "Y"},
+					AdditionalProperties: falseSchema(),
 				},
-				Required:             []string{"X", "Y"},
-				AdditionalProperties: falseSchema(),
 			},
-		},
-		{
-			"nested and embedded",
-			forType[struct {
-				A S
-				S
-			}](),
-			&schema{
-				Type: "object",
-				Properties: map[string]*schema{
-					"A": {
-						Type: "object",
-						Properties: map[string]*schema{
-							"B": {Type: "integer", Description: "bdesc"},
+			{
+				"nested and embedded",
+				forType[struct {
+					A S
+					S
+				}](lax),
+				&schema{
+					Type: "object",
+					Properties: map[string]*schema{
+						"A": {
+							Type: "object",
+							Properties: map[string]*schema{
+								"B": {Type: "integer", Description: "bdesc"},
+							},
+							Required:             []string{"B"},
+							AdditionalProperties: falseSchema(),
 						},
-						Required:             []string{"B"},
-						AdditionalProperties: falseSchema(),
-					},
-					"S": {
-						Type: "object",
-						Properties: map[string]*schema{
-							"B": {Type: "integer", Description: "bdesc"},
+						"S": {
+							Type: "object",
+							Properties: map[string]*schema{
+								"B": {Type: "integer", Description: "bdesc"},
+							},
+							Required:             []string{"B"},
+							AdditionalProperties: falseSchema(),
 						},
-						Required:             []string{"B"},
-						AdditionalProperties: falseSchema(),
 					},
+					Required:             []string{"A", "S"},
+					AdditionalProperties: falseSchema(),
 				},
-				Required:             []string{"A", "S"},
-				AdditionalProperties: falseSchema(),
 			},
-		},
+		}
 	}
 
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			if diff := cmp.Diff(test.want, test.got, cmpopts.IgnoreUnexported(jsonschema.Schema{})); diff != "" {
-				t.Fatalf("ForType mismatch (-want +got):\n%s", diff)
-			}
-			// These schemas should all resolve.
-			if _, err := test.got.Resolve(nil); err != nil {
-				t.Fatalf("Resolving: %v", err)
-			}
-		})
+	run := func(t *testing.T, tt test) {
+		if diff := cmp.Diff(tt.want, tt.got, cmpopts.IgnoreUnexported(jsonschema.Schema{})); diff != "" {
+			t.Fatalf("ForType mismatch (-want +got):\n%s", diff)
+		}
+		// These schemas should all resolve.
+		if _, err := tt.got.Resolve(nil); err != nil {
+			t.Fatalf("Resolving: %v", err)
+		}
 	}
+
+	t.Run("strict", func(t *testing.T) {
+		for _, test := range tests(false) {
+			t.Run(test.name, func(t *testing.T) { run(t, test) })
+		}
+	})
+
+	laxTests := append(tests(true), test{
+		"ignore",
+		forType[struct {
+			A int
+			B map[int]int
+			C func()
+		}](true),
+		&schema{
+			Type: "object",
+			Properties: map[string]*schema{
+				"A": {Type: "integer"},
+			},
+			Required:             []string{"A"},
+			AdditionalProperties: falseSchema(),
+		},
+	})
+	t.Run("lax", func(t *testing.T) {
+		for _, test := range laxTests {
+			t.Run(test.name, func(t *testing.T) { run(t, test) })
+		}
+	})
 }
 
 func forErr[T any]() error {
diff --git a/mcp/event_test.go b/mcp/event_test.go
index 2e2e3d2..9b01555 100644
--- a/mcp/event_test.go
+++ b/mcp/event_test.go
@@ -5,6 +5,9 @@
 package mcp
 
 import (
+	"context"
+	"fmt"
+	"slices"
 	"strings"
 	"testing"
 )
@@ -97,3 +100,167 @@ func TestScanEvents(t *testing.T) {
 		})
 	}
 }
+
+func TestMemoryEventStoreState(t *testing.T) {
+	ctx := context.Background()
+
+	appendEvent := func(s *MemoryEventStore, sess string, str StreamID, data string) {
+		if _, err := s.AppendEvent(ctx, sess, str, []byte(data)); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for _, tt := range []struct {
+		name     string
+		actions  func(*MemoryEventStore)
+		want     string // output of debugString
+		wantSize int    // value of nBytes
+	}{
+		{
+			"appends",
+			func(s *MemoryEventStore) {
+				appendEvent(s, "S1", 1, "d1")
+				appendEvent(s, "S1", 2, "d2")
+				appendEvent(s, "S1", 1, "d3")
+				appendEvent(s, "S2", 8, "d4")
+			},
+			"S1 1 first=0 d1 d3; S1 2 first=0 d2; S2 8 first=0 d4",
+			8,
+		},
+		{
+			"stream close",
+			func(s *MemoryEventStore) {
+				appendEvent(s, "S1", 1, "d1")
+				appendEvent(s, "S1", 2, "d2")
+				appendEvent(s, "S1", 1, "d3")
+				appendEvent(s, "S2", 8, "d4")
+				s.StreamClosed(ctx, "S1", 1)
+			},
+			"S1 2 first=0 d2; S2 8 first=0 d4",
+			4,
+		},
+		{
+			"session close",
+			func(s *MemoryEventStore) {
+				appendEvent(s, "S1", 1, "d1")
+				appendEvent(s, "S1", 2, "d2")
+				appendEvent(s, "S1", 1, "d3")
+				appendEvent(s, "S2", 8, "d4")
+				s.SessionClosed(ctx, "S1")
+			},
+			"S2 8 first=0 d4",
+			2,
+		},
+		{
+			"purge",
+			func(s *MemoryEventStore) {
+				appendEvent(s, "S1", 1, "d1")
+				appendEvent(s, "S1", 2, "d2")
+				appendEvent(s, "S1", 1, "d3")
+				appendEvent(s, "S2", 8, "d4")
+				// We are using 8 bytes (d1,d2, d3, d4).
+				// To purge 6, we remove the first of each stream, leaving only d3.
+				s.SetMaxBytes(2)
+			},
+			// The other streams remain, because we may add to them.
+			"S1 1 first=1 d3; S1 2 first=1; S2 8 first=1",
+			2,
+		},
+		{
+			"purge append",
+			func(s *MemoryEventStore) {
+				appendEvent(s, "S1", 1, "d1")
+				appendEvent(s, "S1", 2, "d2")
+				appendEvent(s, "S1", 1, "d3")
+				appendEvent(s, "S2", 8, "d4")
+				s.SetMaxBytes(2)
+				// Up to here, identical to the "purge" case.
+				// Each of these additions will result in a purge.
+				appendEvent(s, "S1", 2, "d5") // remove d3
+				appendEvent(s, "S1", 2, "d6") // remove d5
+			},
+			"S1 1 first=2; S1 2 first=2 d6; S2 8 first=1",
+			2,
+		},
+		{
+			"purge resize append",
+			func(s *MemoryEventStore) {
+				appendEvent(s, "S1", 1, "d1")
+				appendEvent(s, "S1", 2, "d2")
+				appendEvent(s, "S1", 1, "d3")
+				appendEvent(s, "S2", 8, "d4")
+				s.SetMaxBytes(2)
+				// Up to here, identical to the "purge" case.
+				s.SetMaxBytes(6) // make room
+				appendEvent(s, "S1", 2, "d5")
+				appendEvent(s, "S1", 2, "d6")
+			},
+			// The other streams remain, because we may add to them.
+			"S1 1 first=1 d3; S1 2 first=1 d5 d6; S2 8 first=1",
+			6,
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			s := NewMemoryEventStore(nil)
+			tt.actions(s)
+			got := s.debugString()
+			if got != tt.want {
+				t.Errorf("\ngot  %s\nwant %s", got, tt.want)
+			}
+			if g, w := s.nBytes, tt.wantSize; g != w {
+				t.Errorf("got size %d, want %d", g, w)
+			}
+		})
+	}
+}
+
+func TestMemoryEventStoreAfter(t *testing.T) {
+	ctx := context.Background()
+	s := NewMemoryEventStore(nil)
+	s.SetMaxBytes(4)
+	s.AppendEvent(ctx, "S1", 1, []byte("d1"))
+	s.AppendEvent(ctx, "S1", 1, []byte("d2"))
+	s.AppendEvent(ctx, "S1", 1, []byte("d3"))
+	s.AppendEvent(ctx, "S1", 2, []byte("d4")) // will purge d1
+	want := "S1 1 first=1 d2 d3; S1 2 first=0 d4"
+	if got := s.debugString(); got != want {
+		t.Fatalf("got state %q, want %q", got, want)
+	}
+
+	for _, tt := range []struct {
+		sessionID string
+		streamID  StreamID
+		index     int
+		want      []string
+		wantErr   string // if non-empty, error should contain this string
+	}{
+		{"S1", 1, 0, nil, "purge"},
+		{"S1", 1, 1, []string{"d2", "d3"}, ""},
+		{"S1", 1, 2, []string{"d3"}, ""},
+		{"S1", 2, 0, []string{"d4"}, ""},
+		{"S1", 3, 0, nil, "unknown stream ID"},
+		{"S2", 0, 0, nil, "unknown session ID"},
+	} {
+		t.Run(fmt.Sprintf("%s-%d-%d", tt.sessionID, tt.streamID, tt.index), func(t *testing.T) {
+			var got []string
+			for d, err := range s.After(ctx, tt.sessionID, tt.streamID, tt.index) {
+				if err != nil {
+					if tt.wantErr == "" {
+						t.Fatalf("unexpected error %q", err)
+					} else if g := err.Error(); !strings.Contains(g, tt.wantErr) {
+						t.Fatalf("got error %q, want it to contain %q", g, tt.wantErr)
+					} else {
+						return
+					}
+				}
+				got = append(got, string(d))
+			}
+			if tt.wantErr != "" {
+				t.Fatalf("expected error containing %q, got nil", tt.wantErr)
+			}
+			if !slices.Equal(got, tt.want) {
+				t.Errorf("got %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/mcp/streamable_test.go b/mcp/streamable_test.go
index 05bf59e..87664da 100644
--- a/mcp/streamable_test.go
+++ b/mcp/streamable_test.go
@@ -555,7 +555,7 @@ func mustMarshal(t *testing.T, v any) json.RawMessage {
 
 func TestEventID(t *testing.T) {
 	tests := []struct {
-		sid streamID
+		sid StreamID
 		idx int
 	}{
 		{0, 0},