server

package
v0.3.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 7, 2026 License: Apache-2.0 Imports: 31 Imported by: 0

README

server

MCP server implementation: Dispatcher, transports, middleware, subscriptions.

What belongs here

  • Server struct and options (NewServer, WithBearerToken, WithToolTimeout, etc.)
  • Dispatcher — JSON-RPC routing, method handlers, session state
  • Transports: SSE (transport.go), Streamable HTTP (streamable_transport.go)
  • InProcessTransport — core.Transport implementation for testing/embedding
  • Middleware chain (WithMiddleware, LoggingMiddleware)
  • Server-to-client request infrastructure (sendServerRequest, routeServerResponse)
  • Resource subscriptions (WithSubscriptions, NotifyResourceUpdated)
  • Broadcast notifications (Broadcast) — fan out to all connected sessions
  • Extension registration (WithExtension) — extensions declare capabilities in initialize response
  • Startup validation (validateExtensionRefs) — calls RefValidator on registered extensions to warn about unresolvable resource references

Dependencies

  • core/ — protocol types (Request, Response, ToolDef, etc.)
  • servicekit — SSE hub, graceful shutdown
  • Does NOT import client/

In-process transport

For testing, use NewInProcessTransport(srv) with client.WithTransport():

transport := server.NewInProcessTransport(srv,
    server.WithServerRequestHandler(func(ctx context.Context, req *core.Request) *core.Response {
        return client.HandleServerRequest(req) // for sampling/elicitation
    }),
    server.WithNotificationHandler(func(method string, params []byte) {
        // capture notifications in tests
    }),
)
c := client.NewClient("memory://", info, client.WithTransport(transport))

The in-process transport passes *core.Request/*core.Response directly — no JSON envelope serialization. This catches logic bugs; HTTP transport tests catch wire format bugs.

Documentation

Overview

relay.go — cross-replica notification routing primitives.

In a multi-replica deployment, mcpkit servers need a way for notifications generated on replica K to reach the connected clients on replicas K′. Pattern B is the wiring shape mcpkit recommends: each replica runs a publisher + subscriber pair against a shared pub/sub transport (Redis pubsub, Kafka, NATS, etc.); the publisher pushes notifications outward; the subscriber on every replica receives them and routes into the local per-replica delivery machinery (per-source slots, per-URI subscriptions, session-attached SSE listeners).

Two routing categories cover every notification surface mcpkit emits today:

  • **Capability-shaped** (notifications/tools/list_changed, notifications/resources/list_changed, notifications/prompts/list_changed): no per-subscription filter — every connected client that declared the capability sees the notification. Use CapabilityBroadcastReceiver.

  • **Subscription-shaped** (notifications/events/event, notifications/resources/updated): each replica applies a per-subscription filter (events EventDef.Match per slot, URI prefix per resources/subscribe). Routing happens through domain-specific adapters (events.YieldingSource for events; equivalent for resources/updated when that wire-up lands).

NotificationRelayReceiver is the shared interface a Pattern B transport adapter calls on receive. Self-publish dedup (avoiding a replica re-firing its own publishes) is the transport adapter's internal concern — receivers don't see it.

Source: issue 755.

Package tasks is EXPERIMENTAL and subject to breaking changes.

It implements the MCP Tasks protocol (spec 2025-11-25) as a reusable library on top of mcpkit. Servers register task-capable tools; the library handles protocol methods (tasks/get, tasks/result, tasks/list, tasks/cancel), middleware-based tools/call interception, and in-memory task state management.

Stability: experimental. The Go API will change as the spec evolves and conformance tests are published.

Index

Constants

View Source 
const DefaultStatelessSubscriptionCap = 100

DefaultStatelessSubscriptionCap is the default per-scope concurrent subscriptions/listen stream cap applied when the caller does not override it via WithStatelessSubscriptionCap. Chosen to give public-facing deployments out-of-the-box defense against churn — same posture as net/http.Transport.MaxIdleConns. Pass a negative number to WithStatelessSubscriptionCap to disable the cap entirely.

View Source 
const ErrCodeCancelled = -32800

ErrCodeCancelled is the JSON-RPC error code for a cancelled request.

Variables

View Source 
var ErrStatelessStreamCapExceeded = errors.New("stateless subscription stream cap exceeded")

ErrStatelessStreamCapExceeded indicates the requesting scope has hit the per-scope concurrent-stream cap configured via WithStatelessSubscriptionCap. Returned by the stateless subscription registry's tryRegister so handleStatelessSubscribe can map it to the wire error.

View Source 
var ErrStatelessStreamRateLimited = errors.New("stateless subscription stream rate limited")

ErrStatelessStreamRateLimited indicates the requesting scope has exceeded the per-scope open-rate configured via WithStatelessSubscriptionRateLimit.

View Source 
var ErrSubscriptionCapExceeded = errors.New("subscription cap exceeded")

ErrSubscriptionCapExceeded indicates the calling session has hit the per-session concurrent-subscription cap configured via WithSubscriptionCap. Returned by subscribe so handleResourcesSubscribe can map it to the wire error.

View Source 
var ErrSubscriptionRateLimited = errors.New("subscription rate limited")

ErrSubscriptionRateLimited indicates the calling session has exceeded the per-session subscribe rate configured via WithSubscriptionRateLimit.

Functions

func DefaultStatelessSubscriptionScope added in v0.2.47 

func DefaultStatelessSubscriptionScope(r *http.Request) string

DefaultStatelessSubscriptionScope returns the host portion of r.RemoteAddr — what the OS reports for the TCP peer — and falls back to RemoteAddr verbatim if the port split fails. This is the conservative default: it cannot be spoofed by a malicious client (headers are not consulted), but it lumps every client behind a single reverse-proxy IP into the same scope.

For a deployment where the operator trusts a header like X-Forwarded-For to identify clients, supply a custom StatelessSubscriptionScopeFunc via WithStatelessSubscriptionScope.

Stability: the wire surface this guards (SEP-2575 subscriptions/listen) is in the 2026-07-28 release candidate, not Final. The cap and rate-limit mechanism is mcpkit-internal and not spec-defined — if a future SEP standardizes either, callers may need to migrate.

func IsMethodNotFound added in v0.2.41 

func IsMethodNotFound(err error) bool

IsMethodNotFound reports whether the error is a JSON-RPC method-not-found (-32601).

func RegisterTasksV1 added in v0.2.41 

func RegisterTasksV1(cfg TasksConfigV1)

Register hooks up tasks support on the given server:

  • Installs middleware that intercepts tools/call for task-eligible requests
  • Registers tasks/get, tasks/result, tasks/list, tasks/cancel handlers
  • Advertises the tasks capability in the initialize response

Must be called before accepting connections.

func WithTaskContext added in v0.2.41 

func WithTaskContext(ctx context.Context, tc *TaskContext) context.Context

WithTaskContext returns a context carrying the TaskContext.

Types

type BaseErrorHandler added in v0.1.14 

type BaseErrorHandler struct{}

BaseErrorHandler provides no-op defaults for all ErrorHandler methods. Embed this in your implementation to only override specific methods: 

type MyHandler struct {
    server.BaseErrorHandler
    logger *slog.Logger
}
func (h *MyHandler) OnSessionExpire(id string, err error) {
    h.logger.Error("session expired", "id", id, "err", err)
}

func (BaseErrorHandler) OnKeepaliveFailure added in v0.1.14 

func (BaseErrorHandler) OnKeepaliveFailure(string, int)

func (BaseErrorHandler) OnSessionExpire added in v0.1.14 

func (BaseErrorHandler) OnSessionExpire(string, error)

func (BaseErrorHandler) OnTransportError added in v0.1.14 

func (BaseErrorHandler) OnTransportError(error)

type CapabilityBroadcastReceiver added in v0.2.47 

type CapabilityBroadcastReceiver struct {
	// contains filtered or unexported fields
}

CapabilityBroadcastReceiver is the reference NotificationRelayReceiver for capability-shaped notifications — tools/list_changed, resources/list_changed, prompts/list_changed, and any future notification with the same shape (no per-subscription filter; every connected client with the capability sees it).

On Receive it forwards to Server.Broadcast, which the per-transport session machinery fans out to every connected client on this replica.

func NewCapabilityBroadcastReceiver added in v0.2.47 

func NewCapabilityBroadcastReceiver(srv *Server) *CapabilityBroadcastReceiver

NewCapabilityBroadcastReceiver constructs a CapabilityBroadcastReceiver wired to srv. The transport adapter handles self-publish dedup internally before invoking Receive — adopters don't pass an origin marker here.

func (*CapabilityBroadcastReceiver) Receive added in v0.2.47 

func (r *CapabilityBroadcastReceiver) Receive(_ context.Context, method string, params any)

Receive implements NotificationRelayReceiver. Forwards to Server.BroadcastToSessions (NOT Server.Broadcast) with a background context — calling Broadcast here would re-fire the installed NotificationRelay and loop indefinitely through the Pattern B transport. BroadcastToSessions delivers to local sessions only, which is what the relay receive path needs.

The relay is fire-and-forget at the notification level; the transport's ctx cancellation belongs to the transport loop, not to the broadcast fan-out on this replica.

type Dispatcher 

type Dispatcher struct {
	Reg *Registry
	// contains filtered or unexported fields
}

Dispatcher routes JSON-RPC requests to the appropriate handler.

func NewDispatcher 

func NewDispatcher(info core.ServerInfo) *Dispatcher

NewDispatcher creates a dispatcher with the given server identity.

func (*Dispatcher) Close 

func (d *Dispatcher) Close()

Close tears down all per-session state on the Dispatcher. Transports must call this when a session disconnects (SSE stream closes, DELETE request, client close). Centralizes cleanup so that adding new per-session state (subscriptions, sampling, elicitation) only requires updating this method, not every transport. Safe to call multiple times and on dispatchers with no session state.

func (*Dispatcher) Dispatch 

func (d *Dispatcher) Dispatch(ctx context.Context, req *core.Request) (resp *core.Response)

Dispatch routes a JSON-RPC request and returns the response. Returns nil for notifications (no response expected).

func (*Dispatcher) NegotiatedVersion 

func (d *Dispatcher) NegotiatedVersion() string

NegotiatedVersion returns the protocol version negotiated during initialization.

func (*Dispatcher) RegisterCompletion 

func (d *Dispatcher) RegisterCompletion(refType, name string, handler core.CompletionHandler)

RegisterCompletion registers a completion handler for a specific reference. refType is "ref/prompt" or "ref/resource". name is the prompt name or resource URI template.

func (*Dispatcher) RegisterPrompt 

func (d *Dispatcher) RegisterPrompt(def core.PromptDef, handler core.PromptHandler)

RegisterPrompt adds a prompt to the dispatcher's registry. Panics if any argument Schema is set but invalid — see Dispatcher.RegisterTool for rationale. Use Registry.AddPrompt directly to handle schema errors.

func (*Dispatcher) RegisterResource 

func (d *Dispatcher) RegisterResource(def core.ResourceDef, handler core.ResourceHandler)

RegisterResource adds a resource to the dispatcher's registry.

func (*Dispatcher) RegisterResourceTemplate 

func (d *Dispatcher) RegisterResourceTemplate(def core.ResourceTemplate, handler core.TemplateHandler)

RegisterResourceTemplate adds a URI template resource to the dispatcher's registry.

func (*Dispatcher) RegisterTool 

func (d *Dispatcher) RegisterTool(def core.ToolDef, handler core.ToolHandler)

RegisterTool adds a tool to the dispatcher's registry. Panics if def.InputSchema is set but invalid — schema compilation failures at registration time are programmer errors (the schema is hard-coded in the server binary), so fail-fast is preferred. Use Registry.AddTool directly if you need to handle schema errors programmatically.

func (*Dispatcher) Roots added in v0.1.24 

func (d *Dispatcher) Roots() []core.Root

Roots returns a snapshot of the client's most recently reported roots. The returned slice is safe to retain and mutate — it is a fresh copy. Returns nil if no roots/list fetch has completed yet (the typical state until the client first sends notifications/roots/list_changed).

func (*Dispatcher) RouteResponse 

func (d *Dispatcher) RouteResponse(resp *core.Response) bool

RouteResponse routes an incoming JSON-RPC response from the client to a pending server-to-client request. Returns true if matched, false if no pending request was found for the response ID.

func (*Dispatcher) SetNotifyFunc 

func (d *Dispatcher) SetNotifyFunc(fn core.NotifyFunc)

SetNotifyFunc sets the notification delivery function for this dispatcher. Thread-safe: can be called concurrently with getNotifyFunc reads.

func (*Dispatcher) SetPushRequest added in v0.1.24 

func (d *Dispatcher) SetPushRequest(fn func(json.RawMessage))

SetPushRequest installs the transport's push function for server-initiated JSON-RPC requests (e.g. roots/list, sampling/createMessage). Called once per persistent stream by stdio, in-process, and the SSE/Streamable-HTTP GET SSE handlers; not called by the per-POST request path (which wires its own request-scoped closure). Thread-safe.

type ErrorHandler added in v0.1.14 

type ErrorHandler interface {
	// OnSessionExpire is called when a session is terminated due to idle
	// timeout, keepalive failure, or explicit close.
	OnSessionExpire(sessionID string, reason error)

	// OnTransportError is called for transport-level errors that don't
	// result in a client response (connection drops, body read failures).
	OnTransportError(err error)

	// OnKeepaliveFailure is called each time a keepalive ping fails.
	// consecutiveFailures is the current streak count.
	OnKeepaliveFailure(sessionID string, consecutiveFailures int)
}

ErrorHandler receives out-of-band errors that aren't returned to a specific client request — session lifecycle errors, transport failures, keepalive timeouts. Implement for observability, alerting, or diagnostics.

Embed BaseErrorHandler to only override the methods you care about.

type ExecConfig added in v0.1.16 

type ExecConfig struct {
	// Name is the MCP tool name (required).
	Name string

	// Description is the tool description shown in tools/list.
	Description string

	// Command is the executable path or name (required).
	Command string

	// Args are static arguments prepended to every invocation.
	Args []string

	// Env are additional environment variables in KEY=VALUE format.
	// These are appended to the current process environment.
	Env []string

	// Dir is the working directory for the subprocess.
	// If empty, inherits the current process working directory.
	Dir string

	// Timeout is the maximum execution time per invocation.
	// Zero means no timeout beyond the request context.
	Timeout time.Duration

	// InputSchema is the JSON Schema for the tool's arguments.
	// If nil, defaults to {"type": "object"}.
	InputSchema any

	// BuildArgs transforms the tool request's JSON arguments into CLI args.
	// The returned args are appended after the static Args.
	// If nil, no dynamic arguments are added — only static Args are used.
	BuildArgs func(args json.RawMessage) ([]string, error)
}

ExecConfig configures a ToolExec tool that wraps a CLI binary as an MCP tool.

This is useful for wrapping existing CLI tools (like build systems, linters, code generators) as MCP tools without reimplementing their logic. For example, a presentation tool could expose its build command: 

srv.Register(server.ToolExec(server.ExecConfig{
    Name:    "build_deck",
    Command: "slyds",
    Args:    []string{"build"},
    BuildArgs: func(args json.RawMessage) ([]string, error) {
        var p struct{ Deck string `json:"deck"` }
        json.Unmarshal(args, &p)
        return []string{"--deck", p.Deck}, nil
    },
}))

type HandleStore added in v0.2.47 

type HandleStore[T any] interface {
	// Mint stores v under a freshly-minted opaque id and returns the
	// id. ttl=0 falls back to the store's default TTL (if any); ttl<0
	// forces "never expires" regardless of any default; ttl>0 pins a
	// per-handle override.
	Mint(v T, ttl time.Duration) string

	// Put updates the value stored under an existing id, refreshing
	// its TTL (same ttl semantics as Mint). Returns true when an
	// entry existed under id (now overwritten); false when there
	// was no such entry (in which case Put inserts it under that
	// id — useful for callers that already have a id from elsewhere
	// they want the store to honor, e.g. a deterministic id derived
	// from input).
	//
	// Mint vs Put: use Mint when you want the store to choose an id;
	// use Put to update-in-place after an earlier Mint, or to seed
	// an entry under a caller-chosen id.
	Put(id string, v T, ttl time.Duration) bool

	// Get returns the value stored under id, or zero+false if no such
	// handle is registered OR if the handle has expired. Lazy expiry —
	// implementations are expected to surface ok=false past TTL even
	// without a background sweep.
	Get(id string) (T, bool)

	// Delete removes the entry under id and returns whether it was
	// present. Safe to call with an unknown id (no-op).
	Delete(id string) bool

	// Len returns the current entry count (may include expired-but-
	// not-swept entries depending on the implementation). Primarily
	// for tests and ops introspection.
	Len() int

	// Close releases any background resources (GC goroutines, network
	// connections to backends). Always safe to call; idempotent.
	Close()
}

HandleStore is the typed store for SEP-2567 state handles. One store per logical type (cart, document, transaction, ...) keeps the type assertions in tool handlers clean. A single Server may hold any number of HandleStores side by side; they share no state.

All methods are safe for concurrent use.

func NewHandleStore added in v0.2.47 

func NewHandleStore[T any](opts ...HandleStoreOption) HandleStore[T]

NewHandleStore returns the default in-memory implementation. Most callers should use this; integrators who need cross-replica sharing construct a backend-specific store directly (which returns the same HandleStore[T] interface).

Pass WithHandleGCInterval to enable background expiry sweeps; without it the store relies on lazy expiry (Get checks the timestamp).

type HandleStoreOption added in v0.2.47 

type HandleStoreOption func(*handleStoreOpts)

HandleStoreOption configures an InMemoryHandleStore at construction. Backend-specific implementations may accept their own option types.

func WithHandleDefaultTTL added in v0.2.47 

func WithHandleDefaultTTL(ttl time.Duration) HandleStoreOption

WithHandleDefaultTTL sets the default per-handle TTL applied by Mint when the caller passes a zero ttl. Zero (the default) means "no expiry".

func WithHandleGCInterval added in v0.2.47 

func WithHandleGCInterval(d time.Duration) HandleStoreOption

WithHandleGCInterval enables periodic sweep of expired handles. 0 (the default) skips the background goroutine — callers can still rely on lazy expiry (Get returns ok=false for an expired handle even if the entry hasn't been swept). A small interval keeps memory tight for stores with high turnover; tests pass a few-ms interval to drive the expiry path deterministically.

func WithHandleIDPrefix added in v0.2.47 

func WithHandleIDPrefix(prefix string) HandleStoreOption

WithHandleIDPrefix prepends a short label to every minted handle id. Cosmetic; doesn't affect security or uniqueness. Useful in multi-store deployments so a stray id is greppable to its origin ("cart-...", "doc-...").

type InMemoryHandleStore added in v0.2.47 

type InMemoryHandleStore[T any] struct {
	// contains filtered or unexported fields
}

InMemoryHandleStore is the default HandleStore implementation. Holds entries in a Go map under an RWMutex. No persistence across process restarts — for cross-replica deployments use a backend-backed implementation (tracked by panyam/mcpkit#471).

Exported so tests and code that needs the concrete type (e.g. for custom introspection beyond Len) can construct it directly. New code should prefer NewHandleStore's interface return.

func NewInMemoryHandleStore added in v0.2.47 

func NewInMemoryHandleStore[T any](opts ...HandleStoreOption) *InMemoryHandleStore[T]

NewInMemoryHandleStore constructs the in-memory implementation directly, returning the concrete struct for callers that need it. Most call sites should prefer NewHandleStore (returns interface).

func (*InMemoryHandleStore[T]) Close added in v0.2.47 

func (s *InMemoryHandleStore[T]) Close()

Close stops the background gc goroutine, if any. Idempotent.

func (*InMemoryHandleStore[T]) Delete added in v0.2.47 

func (s *InMemoryHandleStore[T]) Delete(id string) bool

Delete removes the entry under id and returns whether it was present. Safe to call with an unknown id (no-op).

func (*InMemoryHandleStore[T]) Get added in v0.2.47 

func (s *InMemoryHandleStore[T]) Get(id string) (T, bool)

Get returns the value stored under id, or zero value + false if no such handle is registered OR if the handle has expired. Lazy expiry — no background goroutine required to enforce TTL on the read path.

Note: a handle that expires between Get's read-lock and the caller's next operation will still appear valid to the caller. Tool handlers that absolutely must operate atomically should arrange transactional locking outside this store (or use a persistent backend with native atomic ops — see panyam/mcpkit#471).

func (*InMemoryHandleStore[T]) Len added in v0.2.47 

func (s *InMemoryHandleStore[T]) Len() int

Len returns the current entry count (including expired-but-not-swept).

func (*InMemoryHandleStore[T]) Mint added in v0.2.47 

func (s *InMemoryHandleStore[T]) Mint(v T, ttl time.Duration) string

Mint stores v under a freshly-minted opaque id and returns the id.

ttl overrides the store's defaultTTL for this entry; pass 0 to fall back to the default. Pass a negative ttl to force "no expiry" even when the store has a non-zero default.

IDs are 128 bits of crypto-random base32 (26 chars), optionally prefixed via WithHandleIDPrefix.

func (*InMemoryHandleStore[T]) Put added in v0.2.47 

func (s *InMemoryHandleStore[T]) Put(id string, v T, ttl time.Duration) bool

Put updates the entry under id with v and refreshes its TTL using the same semantics as Mint. Returns true when an entry existed under id (now overwritten); false when there was no such entry (Put inserts it under that id either way).

type InMemoryMessageQueue added in v0.2.41 

type InMemoryMessageQueue struct {
	// contains filtered or unexported fields
}

InMemoryMessageQueue is an in-memory TaskMessageQueue implementation.

func NewInMemoryMessageQueue added in v0.2.41 

func NewInMemoryMessageQueue() *InMemoryMessageQueue

NewInMemoryMessageQueue creates a new in-memory message queue.

func (*InMemoryMessageQueue) Cleanup added in v0.2.41 

func (q *InMemoryMessageQueue) Cleanup()

func (*InMemoryMessageQueue) Dequeue added in v0.2.41 

func (q *InMemoryMessageQueue) Dequeue(taskID string) (QueuedMessage, bool)

func (*InMemoryMessageQueue) DequeueAll added in v0.2.41 

func (q *InMemoryMessageQueue) DequeueAll(taskID string) []QueuedMessage

func (*InMemoryMessageQueue) Enqueue added in v0.2.41 

func (q *InMemoryMessageQueue) Enqueue(taskID string, msg QueuedMessage, maxSize int) error

func (*InMemoryMessageQueue) WaitForMessage added in v0.2.41 

func (q *InMemoryMessageQueue) WaitForMessage(taskID string, done <-chan struct{}) bool

type InMemoryTaskStore added in v0.2.41 

type InMemoryTaskStore struct {
	// contains filtered or unexported fields
}

InMemoryTaskStore is a TaskStore backed by an in-memory map with insertion-ordered keys for cursor pagination.

func NewInMemoryStore added in v0.2.41 

func NewInMemoryStore() *InMemoryTaskStore

NewInMemoryStore creates a new in-memory task store.

func (*InMemoryTaskStore) Cancel added in v0.2.41 

func (s *InMemoryTaskStore) Cancel(taskID, sessionID string) (core.TaskInfo, error)

func (*InMemoryTaskStore) Cleanup added in v0.2.41 

func (s *InMemoryTaskStore) Cleanup()

Cleanup removes all tasks and stops all TTL timers. Used for graceful shutdown and testing.

func (*InMemoryTaskStore) Create added in v0.2.41 

func (s *InMemoryTaskStore) Create(info core.TaskInfo, sessionID string) error

func (*InMemoryTaskStore) Get added in v0.2.41 

func (s *InMemoryTaskStore) Get(taskID, sessionID string) (core.TaskInfo, bool)

func (*InMemoryTaskStore) GetResult added in v0.2.41 

func (s *InMemoryTaskStore) GetResult(taskID, sessionID string) (core.ToolResult, bool)

func (*InMemoryTaskStore) List added in v0.2.41 

func (s *InMemoryTaskStore) List(cursor string, limit int, sessionID string) ([]core.TaskInfo, string)

func (*InMemoryTaskStore) SetResult added in v0.2.41 

func (s *InMemoryTaskStore) SetResult(taskID, sessionID string, result core.ToolResult) error

func (*InMemoryTaskStore) StoreTerminalResult added in v0.2.41 

func (s *InMemoryTaskStore) StoreTerminalResult(taskID, sessionID string, status core.TaskStatus, result core.ToolResult, statusMsg string) error

StoreTerminalResult atomically stores the result and transitions to a terminal status. Rejects if the task is already terminal (prevents cancel→completed races and double-completion).

func (*InMemoryTaskStore) Update added in v0.2.41 

func (s *InMemoryTaskStore) Update(taskID, sessionID string, fn func(*core.TaskInfo)) error

func (*InMemoryTaskStore) WaitForResult added in v0.2.41 

func (s *InMemoryTaskStore) WaitForResult(ctx context.Context, taskID, sessionID string) (core.ToolResult, core.TaskInfo, error)

func (*InMemoryTaskStore) WaitForUpdate added in v0.2.41 

func (s *InMemoryTaskStore) WaitForUpdate(ctx context.Context, taskID, sessionID string) error

type InProcessOption 

type InProcessOption func(*InProcessTransport)

InProcessOption configures an InProcessTransport.

func WithNotificationHandler 

func WithNotificationHandler(h core.NotificationHandler) InProcessOption

WithNotificationHandler sets a callback for server-to-client notifications (logging, progress, resource updates). Useful in tests to verify notification delivery.

func WithServerRequestHandler 

func WithServerRequestHandler(h core.ServerRequestHandler) InProcessOption

WithServerRequestHandler sets a handler for server-to-client requests (sampling/createMessage, elicitation/create). When the server sends a request during tool execution, this handler is called to produce a response.

type InProcessTransport 

type InProcessTransport struct {
	// contains filtered or unexported fields
}

InProcessTransport implements core.Transport by dispatching directly to a Server in the same process. No HTTP, no serialization of the Request/Response envelope — domain params (json.RawMessage) pass through as-is.

func NewInProcessTransport 

func NewInProcessTransport(srv *Server, opts ...InProcessOption) *InProcessTransport

NewInProcessTransport creates a transport that dispatches to the given server in-memory. Use with client.WithTransport() to create a client that talks to the server without HTTP.

Example: 

srv := server.New(core.ServerInfo{Name: "test", Version: "1.0"})
srv.RegisterTool(def, handler)
transport := server.NewInProcessTransport(srv,
    server.WithServerRequestHandler(mySamplingHandler),
)
c := client.New("memory://", core.ClientInfo{...}, client.WithTransport(transport))

func (*InProcessTransport) Call 

func (t *InProcessTransport) Call(ctx context.Context, req *core.Request) (*core.Response, error)

Call dispatches a JSON-RPC request and returns the response. Passes *Request directly to the server — no HTTP marshal/unmarshal overhead.

func (*InProcessTransport) Close 

func (t *InProcessTransport) Close() error

Close tears down the session.

func (*InProcessTransport) Connect 

func (t *InProcessTransport) Connect(ctx context.Context) error

Connect creates a per-session dispatcher and wires notification delivery and server-to-client request handling.

func (*InProcessTransport) Notify 

func (t *InProcessTransport) Notify(ctx context.Context, req *core.Request) error

Notify dispatches a JSON-RPC notification (no response expected).

func (*InProcessTransport) SessionID 

func (t *InProcessTransport) SessionID() string

SessionID returns "memory" for the in-process transport.

type MethodHandler added in v0.2.32 

type MethodHandler func(ctx core.MethodContext, id json.RawMessage, params json.RawMessage) *core.Response

MethodHandler handles a custom JSON-RPC method registered via WithMethodHandler or Server.HandleMethod. The handler receives a typed MethodContext with full session capabilities (EmitLog, Sample, Elicit, AuthClaims, Notify, etc.), the request ID, and raw JSON params.

Return a *core.Response with the result or error. Use core.NewResponse for success, core.NewErrorResponse for errors.

Custom methods participate in middleware and require initialization (they're dispatched after the "initialized" gate, same as tools/resources).

type Middleware 

type Middleware func(ctx context.Context, req *core.Request, next MiddlewareFunc) (*core.Response, error)

Middleware intercepts a JSON-RPC request. The middleware can:

  • Call next to continue the chain and return its response
  • Return a *core.Response to short-circuit with a JSON-RPC error
  • Return a non-nil error to short-circuit at the transport layer (e.g., a *core.AuthError with WWW-Authenticate triggers an HTTP 403/401)

Convention: when to use which return path

There is no programmatic distinction between transport-level and protocol-level errors — it is purely a contract between the middleware author and the transport layer:

  • Response.Error: JSON-RPC protocol errors (parse error, invalid params, method not found). Client receives a normal JSON-RPC error reply.

  • Response.Result with IsError: tool-level errors (the tool ran and produced an error result). Client receives tools/call with isError: true.

  • error return: transport-level short-circuit (auth challenges, scope step-up, rate limits requiring Retry-After). Transport layer maps to an HTTP response (writeAuthError handles *core.AuthError → 401/403).

Receiving middleware should not differentiate — just propagate (resp, err) from next() as-is, with whatever inspection logic it needs around it. Only the transport layer consumes the error.

Middleware sees the full request (method, params, ID) and the context (which includes auth claims via core.AuthClaims(ctx) and session notification state). The response from next can be inspected or modified before returning.

Example — logging middleware: 

mcpkit.WithMiddleware(mcpkit.LoggingMiddleware(logger))

Example — per-method rate limiting: 

func RateLimitMiddleware(limiter *rate.Limiter) mcpkit.Middleware {
    return func(ctx context.Context, req *mcpkit.Request, next mcpkit.MiddlewareFunc) (*mcpkit.Response, error) {
        if !limiter.Allow() {
            return mcpkit.NewErrorResponse(req.ID, -32000, "rate limit exceeded"), nil
        }
        return next(ctx, req)
    }
}

func LoggingMiddleware 

func LoggingMiddleware(logger *log.Logger) Middleware

LoggingMiddleware logs every JSON-RPC request with method name, latency, and error status. Useful for debugging and operational monitoring.

Example output: 

MCP initialize ok [1.2ms]
MCP tools/call ok [45.3ms]
MCP tools/call error=-32602 (invalid params) [0.1ms]

func ToolCallLogger added in v0.2.41 

func ToolCallLogger(logger *log.Logger) Middleware

ToolCallLogger logs tools/call requests with the tool name and isError status. Complements LoggingMiddleware: that one only sees JSON-RPC errors, while this one also surfaces "in-stream" tool errors — tool results with isError: true (returned at the JSON-RPC layer as success). Useful for visibility into authorization denials, scope failures, and other tool-level error signals.

Non-tools/call requests pass through without logging.

Example output: 

tool=read_document ok
tool=update_document isError=true text="{\"error\":\"insufficient_scope\",..."
tool=initiate_payment isError=true text="..."

type MiddlewareFunc 

type MiddlewareFunc func(context.Context, *core.Request) (*core.Response, error)

MiddlewareFunc is the signature for the next handler in the middleware chain.

type NotificationRelay added in v0.2.47 

type NotificationRelay interface {
	Publish(ctx context.Context, method string, params any)
}

NotificationRelay is the publish side of a Pattern B transport for capability-shaped notifications (notifications/tools/list_changed, notifications/resources/list_changed, etc.). Adopters install one via WithNotificationRelay; Server.Broadcast then fires Publish(ctx, method, params) before the local BroadcastToSessions fan-out so the same notification reaches every connected client across every replica.

Publish is fire-and-forget — Server.Broadcast does not surface errors. Implementations log internally if a publish fails; the local BroadcastToSessions fan-out still runs regardless.

Concurrency: Publish may be invoked from any goroutine that calls Server.Broadcast (handlers, registry.OnChange, etc.). Implementations must be safe for concurrent calls.

The receive-side wiring (a separate subscriber loop calling Server.BroadcastToSessions on cross-replica receives) is the transport adapter's responsibility. The reference Redis implementation is redisstore.CapabilityBus, which satisfies NotificationRelay on the publish side AND drives BroadcastToSessions on the subscribe side via a NotificationRelayReceiver wired internally.

type NotificationRelayReceiver added in v0.2.47 

type NotificationRelayReceiver interface {
	Receive(ctx context.Context, method string, params any)
}

NotificationRelayReceiver is the callback shape mcpkit expects from a Pattern B transport's subscriber side. The transport adapter calls Receive once per cross-replica notification destined for this replica (i.e. after the transport's own self-publish filtering).

Implementations are domain-specific:

  • CapabilityBroadcastReceiver forwards to Server.Broadcast for catalog-mutation notifications.
  • events.YieldingSource routes via per-slot fanout so per- subscription Match / Transform run.
  • ResourcesUpdatedReceiver routes via the per-URI subscription set with prefix match.

Concurrency: Receive may be invoked from any goroutine (typically the transport's receive loop). Implementations must be safe for concurrent calls.

type NotificationRouter added in v0.2.47 

type NotificationRouter struct {
	// contains filtered or unexported fields
}

NotificationRouter routes received notifications by method name to per-method NotificationRelayReceivers. Adopters configure one NotificationRouter as the NotificationRelay's receiver, registering per-method handlers (CapabilityBroadcastReceiver for catalog notifications, ResourcesUpdatedReceiver for subscription-shaped resources/updated, etc.). Methods without a registered handler are dropped silently — appropriate for shared Pattern B transports where not every replica subscribes to every method.

Concurrency: Handle and Receive are safe for concurrent use. Routing decisions snapshot the dispatch table under a read lock so in-flight notifications don't race with late registrations.

func NewNotificationRouter added in v0.2.47 

func NewNotificationRouter() *NotificationRouter

NewNotificationRouter constructs an empty router. Register per-method handlers via Handle before installing the router on a NotificationRelay's receiver slot.

func (*NotificationRouter) Handle added in v0.2.47 

func (m *NotificationRouter) Handle(method string, receiver NotificationRelayReceiver) *NotificationRouter

Handle registers receiver as the handler for the given method. Returns the router for chaining. A subsequent Handle call for the same method replaces the previous handler.

func (*NotificationRouter) Receive added in v0.2.47 

func (m *NotificationRouter) Receive(ctx context.Context, method string, params any)

Receive implements NotificationRelayReceiver. Looks up the per-method handler and forwards. Unknown methods are silently dropped.

type NotifyInterceptor added in v0.2.29 

type NotifyInterceptor func(method string, params any, next core.NotifyFunc)

NotifyInterceptor wraps outgoing server-to-client notifications before they reach the transport. Interceptors see the method and params of every notification (logging, progress, resource updates, custom). Call next to continue the chain, or return without calling next to suppress.

Example — log all outgoing notifications: 

server.WithNotifyInterceptor(func(method string, params any, next core.NotifyFunc) {
    log.Printf("→ notify %s", method)
    next(method, params)
})

type Option 

type Option func(*serverOptions)

Option configures a Server.

func WithAllowLegacyOnDraft added in v0.2.47 

func WithAllowLegacyOnDraft() Option

WithAllowLegacyOnDraft is an opt-in back-compat escape hatch for the SEP-2575 enforcement on 2026-07-28. When set, the legacy session wire (initialize + Mcp-Session-Id) is accepted on the draft protocol version without enforcing the per-request _meta envelope on follow-up requests.

Default (option NOT set): the dispatcher enforces SEP-2575 strictly — on 2026-07-28, every post-initialize request MUST carry `params._meta.io.modelcontextprotocol/{protocolVersion, clientInfo, clientCapabilities}`; missing _meta is rejected with -32602.

Use this only if you have legacy clients pinned to 2026-07-28 that haven't migrated to per-request metadata yet. New servers should leave this off so non-conformant clients fail loudly.

func WithAllowReinitialize added in v0.3.1 

func WithAllowReinitialize() Option

WithAllowReinitialize opts into accepting a second initialize on an already-negotiated session (protocol re-negotiation).

Default (option NOT set): once a session has negotiated a protocol version, a duplicate initialize is rejected with -32600 Invalid Request and the existing session state (negotiated version, client capabilities, client identity) is preserved. This stops a misbehaving or hostile client from rewriting session state mid-flight — downgrading the negotiated version or changing the advertised client identity (issue 421).

Set this only if your deployment genuinely re-negotiates the protocol on a live session.

func WithAllowedRoots deprecated 

func WithAllowedRoots(roots ...string) Option

WithAllowedRoots restricts tool cwd to the given directory prefixes.

Deprecated: per SEP-2577, scheduled for removal in v0.4. See docs/SEP_2577_DEPRECATIONS.md.

func WithAuth 

func WithAuth(v core.AuthValidator) Option

WithAuth sets a custom auth validator (e.g. JWT via mcpkit/auth).

func WithBearerToken 

func WithBearerToken(token string) Option

WithBearerToken sets a static bearer token for authentication. Uses constant-time comparison to prevent timing attacks.

func WithContentChunkMethod added in v0.1.17 

func WithContentChunkMethod(method string) Option

WithContentChunkMethod sets a custom notification method name for streaming tool content chunks. When not set, defaults to core.DefaultContentChunkMethod ("notifications/tools/content_chunk"). Use this if clients expect a different method name for streaming content delivery.

func WithErrorHandler added in v0.1.14 

func WithErrorHandler(h ErrorHandler) Option

WithErrorHandler sets a callback for out-of-band errors (session lifecycle, transport failures, keepalive timeouts). When not set, these errors are only logged via log.Printf.

func WithExtension 

func WithExtension(ext core.ExtensionProvider) Option

WithExtension registers a protocol extension that will be advertised in the initialize response. Extensions declare their ID, spec version, and stability level.

func WithFileInputValidation added in v0.2.44 

func WithFileInputValidation() Option

WithFileInputValidation enables SEP-2356 server-side validation of file-typed tool arguments. When enabled, the dispatcher walks each tool's inputSchema for `x-mcp-file` properties (single string/uri or array-items shape) and runs `core.ValidateFileInput` on every matching argument BEFORE the handler is invoked.

Failures surface as JSON-RPC -32602 with structured `data`: 

{
  "code": -32602,
  "message": "file input \"image\" exceeds size limit",
  "data": {
    "reason": "file_too_large",
    "field": "image",
    "actualSize": 5243904,
    "maxSize": 5242880
  }
}

The wire shape is frozen by panyam/mcpconformance `pending` (`src/scenarios/server/file-inputs/`) (the cross-impl contract). Disabled by default — handlers that prefer to validate themselves can leave it off.

Usage: 

srv := server.NewServer(info,
    server.WithFileInputValidation(),
)

func WithHTTPHandler added in v0.2.32 

func WithHTTPHandler(pattern string, h http.Handler) Option

WithMethodHandler registers a handler for a custom JSON-RPC method. Custom methods are dispatched after initialization (same as tools/resources) and participate in server middleware.

Panics if the method name is a built-in MCP spec method.

Example: 

srv := server.NewServer(info,
    server.WithMethodHandler("events/poll", func(ctx core.MethodContext, id json.RawMessage, params json.RawMessage) *core.Response {
        return core.NewResponse(id, map[string]any{"events": []any{}})
    }),
)

WithHTTPHandler registers a custom HTTP handler on the server's mux. Use this for endpoints that need raw HTTP access (SSE streams, webhooks, file uploads) rather than JSON-RPC request/response.

The pattern follows http.ServeMux conventions (e.g., "/events/stream", "GET /health"). The handler runs alongside MCP transport handlers on the same server.

Example: 

srv := server.NewServer(info,
    server.WithHTTPHandler("GET /events/stream", sseStreamHandler),
    server.WithHTTPHandler("/webhooks/", webhookHandler),
)

func WithListCacheControl added in v0.2.47 

func WithListCacheControl(ttlMs int, scope string) Option

WithListCacheControl configures both SEP-2549 list cache hints in a single call: ttlMs (see WithListTTLMs for value semantics) and cacheScope (core.CacheScopePublic or core.CacheScopePrivate; "" omits the field, which clients default to "public"). The hints apply uniformly to all four list endpoints. Use WithListTTLMs when only the TTL is needed.

func WithListTTLMs added in v0.2.47 

func WithListTTLMs(ms int) Option

WithListTTLMs configures the SEP-2549 cache-freshness hint, in integer milliseconds, attached to every tools/list, prompts/list, resources/list, and resources/templates/list response. The hint tells clients how long they MAY serve a cached copy of the list before re-fetching:

  • Negative values are treated as "no hint" (the wire field is omitted), so clients fall back to notifications/list_changed or their own heuristics. This is also the default when the option is not set.
  • 0 sends an explicit `"ttlMs": 0` — the response SHOULD be considered immediately stale; clients MAY re-fetch every time the list is needed.
  • Positive values send `"ttlMs": N` meaning "fresh for N milliseconds"; clients SHOULD NOT re-fetch before it expires unless they receive a list_changed notification.

The hint applies uniformly to all four list endpoints. To also set the SEP-2549 cacheScope in one call, use WithListCacheControl.

SEP-2549's final review renamed this option's predecessor WithListTTL (integer seconds) to milliseconds. See docs/LIST_TTL_MIGRATION.md.

func WithListen 

func WithListen(addr string) Option

WithListen sets the HTTP listen address.

func WithMeterProvider added in v0.2.47 

func WithMeterProvider(mp core.MeterProvider) Option

WithMeterProvider registers a core.MeterProvider that wraps every dispatched JSON-RPC request with canonical MCP metrics emission. Pair with WithTracerProvider to feed both Tempo and Mimir from a single ext/otel adapter (see commonotel.SetupMetrics — issue 668).

When nil or core.NoopMeterProvider{} is passed, no metrics middleware is installed — zero overhead on the default path. The default (no Option set) is identical to passing Noop.

The metrics middleware is installed one layer INSIDE the SEP-414 trace middleware so the `mcp.tool.duration` histogram captures handler-attributable latency, not span-emission overhead. The trace span still wraps the metrics middleware so any overhead is visible in Tempo.

Active-session counting (mcp.sessions.active) is wired into the streamable HTTP transport at session-create / session-expire boundaries — see Server.RecordSessionDelta. Stateless wire has no sessions so the gauge stays at zero on stateless-only deployments (correct).

func WithMethodHandler added in v0.2.32 

func WithMethodHandler(method string, h MethodHandler) Option

func WithMiddleware 

func WithMiddleware(mw ...Middleware) Option

WithMiddleware registers server-side middleware that intercepts all JSON-RPC requests. Middleware executes in registration order: the first registered middleware is the outermost (runs first on request, last on response).

Middleware runs after auth checks (claims are in context) but before method routing and dispatch.

func WithNotificationRelay added in v0.2.47 

func WithNotificationRelay(relay NotificationRelay) Option

WithNotificationRelay installs a NotificationRelay onto the Server. Server.Broadcast then fires Publish on the relay before running BroadcastToSessions locally — every capability-shaped notification (tools/list_changed, resources/list_changed, prompts/list_changed, application-level broadcasts) reaches connected clients across every replica in the deployment.

Pair with a transport adapter's subscribe-side wiring that calls Server.BroadcastToSessions on cross-replica receives (the reference is redisstore.CapabilityBus, which provides both ends).

Pass nil to disable the relay (default) — Server.Broadcast then fires local-only, matching pre-Pattern-B behavior.

func WithNotifyInterceptor added in v0.2.29 

func WithNotifyInterceptor(fn ...NotifyInterceptor) Option

WithNotifyInterceptor registers interceptors for outgoing notifications. Interceptors execute in registration order (first = outermost).

func WithOnRootsChanged added in v0.1.18 

func WithOnRootsChanged(fn func([]core.Root)) Option

WithOnRootsChanged sets a callback invoked when the client sends notifications/roots/list_changed. The callback receives the root list (which may be empty if the client hasn't provided roots yet). Use this to dynamically update allowed roots or trigger re-validation.

func WithPublicMethods added in v0.2.31 

func WithPublicMethods(methods ...string) Option

WithPublicMethods declares JSON-RPC methods that bypass auth and are accessible without a valid token. Use this for pre-auth capability discovery — clients can call these methods to learn what the server offers before deciding to authenticate.

Default: none (all methods require auth when WithAuth is configured).

Recommended set for discovery: 

server.WithPublicMethods("initialize", "notifications/initialized",
    "tools/list", "resources/list", "resources/templates/list",
    "prompts/list", "ping")

func WithReadResourceCacheControl added in v0.2.47 

func WithReadResourceCacheControl(ttlMs int, scope string) Option

WithReadResourceCacheControl sets the default SEP-2549 cache hints for resources/read responses: ttlMs (integer milliseconds; negative omits the field) and cacheScope (core.CacheScopePublic or CacheScopePrivate).

A resource (or template) handler MAY override either hint per-read by setting core.ResourceResult.TTLMs / .CacheScope on its return value; the server applies these defaults only to fields the handler left unset.

resources/read responses frequently depend on the authenticated user — pass core.CacheScopePrivate unless the content is identical for every caller. See docs/LIST_TTL_MIGRATION.md for the security rationale.

func WithRequestInterceptor added in v0.2.29 

func WithRequestInterceptor(fn ...RequestInterceptor) Option

WithRequestInterceptor registers interceptors for outgoing server-to-client requests (sampling, elicitation). Interceptors execute in registration order.

func WithRequestLogging 

func WithRequestLogging(logger *log.Logger) Option

WithRequestLogging enables HTTP-level request/response logging on the server. Logs every incoming HTTP request with method, path, headers (Mcp-Session-Id, Accept, Authorization presence), and the response status code and content-type. This is transport-level logging — for JSON-RPC dispatch-level logging, use WithMiddleware(LoggingMiddleware(logger)).

Example: 

srv := mcpkit.NewServer(info, mcpkit.WithRequestLogging(log.Default()))

func WithRequestStateSigning added in v0.2.41 

func WithRequestStateSigning(key []byte, ttl time.Duration) Option

WithRequestStateSigning configures the HMAC-SHA256 key the server uses to sign and verify SEP-2322 requestState tokens. The same key is shared by ephemeral MRTR (tools/call InputRequiredResult round-trips) and SEP-2663 Tasks (tasks/get / tasks/update / tasks/cancel) so production deployments only configure HMAC once and have all signed-state surfaces work.

Servers without this option run in unsigned mode:

  • MRTR requestState is a random nonce with no integrity guarantee
  • Tasks requestState is the bare taskID (legacy plaintext mode)

SEP-2322 / SEP-2663 say servers MUST treat requestState as attacker-controlled, so production deployments SHOULD configure a key.

ttl bounds how long a minted token stays valid; zero falls back to 24h.

Per-RegisterTasks override: TasksConfig.RequestStateKey still wins when set explicitly, so callers that need separate keys for tasks vs MRTR (key rotation, security boundaries) can opt out of the shared default.

func WithRootsFetchTimeout deprecated added in v0.1.26 

func WithRootsFetchTimeout(d time.Duration) Option

WithRootsFetchTimeout sets the deadline for server-to-client roots/list requests issued after notifications/roots/list_changed. Default is 30s. Decrease for aggressive fail-fast; increase for slow clients with large root enumerations (monorepos, network mounts). Issue #198.

Deprecated: per SEP-2577, scheduled for removal in v0.4. See docs/SEP_2577_DEPRECATIONS.md.

func WithSchemaValidation added in v0.1.22 

func WithSchemaValidation(enabled bool) Option

WithSchemaValidation toggles call-time JSON Schema validation of tool arguments and prompt arguments against their declared schemas. When enabled (the default), the dispatcher validates incoming arguments before the handler is invoked and returns -32602 Invalid Params with structured error data on failure. When disabled, handlers receive arguments unchecked and are responsible for validation themselves.

Registration-time schema compilation is not affected by this option — malformed schemas still fail fast at RegisterTool/RegisterPrompt time. This option only controls whether the compiled schemas are applied to incoming requests.

Example — opt out of validation: 

srv := mcpkit.NewServer(info, mcpkit.WithSchemaValidation(false))

func WithStatelessSubscriptionCap added in v0.2.47 

func WithStatelessSubscriptionCap(n int) Option

WithStatelessSubscriptionCap sets the maximum number of concurrent SEP-2575 subscriptions/listen streams the transport will accept from a single scope (typically a remote host, configurable via WithStatelessSubscriptionScope). Once the scope is at the cap, additional subscriptions/listen requests fail with core.ErrCodeSubscriptionLimitExceeded (-32010) before the SSE stream is opened. Closing a stream frees a slot for the same scope.

Defaults to DefaultStatelessSubscriptionCap (100) when the option is not called — public-facing deployments are protected out of the box. Pass n > 0 to override, or n < 0 to disable the cap entirely. (n == 0 is treated as "use default" so that the zero value of serverOptions picks up the default.)

Stability: the wire surface this guards (SEP-2575 subscriptions/listen) is in the 2026-07-28 release candidate, not Final. The cap is mcpkit-internal and not spec-defined; if a future SEP standardizes subscription caps, callers may need to migrate. See docs/SEP_2640_STATELESS_INTERACTION.md.

func WithStatelessSubscriptionRateLimit added in v0.2.47 

func WithStatelessSubscriptionRateLimit(r rate.Limit, burst int) Option

WithStatelessSubscriptionRateLimit bounds how fast a single scope may open new subscriptions/listen streams. Implemented as a per-scope token bucket with refill rate r and burst b. Exceeded calls fail with core.ErrCodeSubscriptionLimitExceeded (-32010) carrying reason "rate_limited" in error.data.

Off by default (r <= 0 means unlimited). Disconnects are not metered; the cost being bounded is registration churn, not normal cleanup.

Stability: see WithStatelessSubscriptionCap.

func WithStatelessSubscriptionScope added in v0.2.47 

func WithStatelessSubscriptionScope(fn StatelessSubscriptionScopeFunc) Option

WithStatelessSubscriptionScope overrides the scope-key extractor used by WithStatelessSubscriptionCap and WithStatelessSubscriptionRateLimit. The default (DefaultStatelessSubscriptionScope) returns the host portion of r.RemoteAddr, which is correct for deployments where every client has a distinct OS-visible source IP, but lumps all clients behind a shared reverse proxy into one bucket.

Override only when the operator vouches for the proxy chain: an X-Forwarded-For-aware extractor lets a malicious client spoof a scope key. Document the trust assumption in the deployment.

func WithSubscriptionCap added in v0.2.47 

func WithSubscriptionCap(n int) Option

WithSubscriptionCap sets the maximum number of concurrent resources/subscribe registrations a single session may hold. Once the session is at the cap, additional resources/subscribe calls fail with core.ErrCodeSubscriptionLimitExceeded (-32010). Unsubscribing frees a slot for the same session.

Off by default (n <= 0 means unlimited). Recommended for any public-facing deployment; 100 is a reasonable starting point. The cap is per-session, not global — two sessions can each hold n subscriptions simultaneously.

func WithSubscriptionRateLimit added in v0.2.47 

func WithSubscriptionRateLimit(r rate.Limit, burst int) Option

WithSubscriptionRateLimit bounds how fast a single session may issue resources/subscribe calls. Implemented as a per-session token bucket with refill rate r and burst b. burst is the largest spike the bucket will admit before throttling kicks in; the steady-state ceiling is r subscribes per second per session. Exceeded calls fail with core.ErrCodeSubscriptionLimitExceeded (-32010) carrying reason "rate_limited".

Off by default (r <= 0 means unlimited). unsubscribe is not metered — the goal is to bound the cost of registration churn, not to penalize cleanup.

func WithSubscriptionRejectHook added in v0.2.47 

func WithSubscriptionRejectHook(fn SubscriptionRejectFunc) Option

WithSubscriptionRejectHook installs a callback fired every time a resources/subscribe is refused by WithSubscriptionCap or WithSubscriptionRateLimit, and every time a SEP-2575 subscriptions/listen stream is refused by WithStatelessSubscriptionCap or WithStatelessSubscriptionRateLimit. One callback covers both wires so operators only need to wire observability once.

Argument semantics:

  • On the legacy wire: arg1 is the sessionID, arg2 is the resource URI the client tried to subscribe to.
  • On the stateless wire: arg1 is the scope key returned by the configured StatelessSubscriptionScopeFunc (host of RemoteAddr by default), arg2 is the literal string "subscriptions/listen".

reason is "cap_exceeded" or "rate_limited" on both wires.

func WithSubscriptions 

func WithSubscriptions() Option

WithSubscriptions enables resource subscription support (resources/subscribe, resources/unsubscribe, and notifications/resources/updated). When enabled, the server advertises "subscribe": true in the resources capability and accepts subscription requests from clients.

Use Server.NotifyResourceUpdated(uri) to push change notifications to all sessions that have subscribed to the given URI.

For public-facing deployments also pass WithSubscriptionCap (and optionally WithSubscriptionRateLimit); without those a misbehaving client can subscribe unboundedly and exhaust server resources.

func WithTaskBucketKeyer added in v0.3.1 

func WithTaskBucketKeyer(keyer core.TaskBucketKeyer) Option

WithTaskBucketKeyer sets how the task store isolates tasks per request (issue 485). The keyer maps a request context to the bucket key used as the store's sessionID argument for Create / Get / Update / Cancel / List.

Default (option NOT set): the bucket is the transport session ID. On the legacy wire that is the per-connection session; on the SEP-2575 stateless wire it is "" (no session), so all stateless tasks share one bucket per process — fine for single-tenant deployments, a cross-tenant isolation hole for multi-tenant ones.

Multi-tenant stateless deployments install a keyer that derives the bucket from an authenticated subject so tenants cannot see each other's tasks, e.g.: 

server.WithTaskBucketKeyer(func(ctx context.Context) string {
    return auth.SubjectFromContext(ctx) // your auth accessor
})

The keyer takes a raw context.Context, so mcpkit never imports ext/auth — the coupling to auth is code the deployer writes. Applies to both the v1 (RegisterTasksV1) and v2 (ext/tasks) surfaces and both wires.

func WithToolTimeout 

func WithToolTimeout(d time.Duration) Option

WithToolTimeout sets the maximum duration for tool execution.

func WithTracerProvider added in v0.2.47 

func WithTracerProvider(tp core.TracerProvider) Option

WithTracerProvider registers a core.TracerProvider that wraps every dispatched JSON-RPC request in an inbound span and propagates the W3C Trace Context on outbound notifications and server-to-client requests.

When nil or core.NoopTracerProvider{} is passed, no trace middleware is installed and no outbound _meta injection occurs — the server stays on the zero-overhead default. The default (no Option set) is the same "tracing disabled" behavior.

The trace middleware is positioned OUTERMOST in the user middleware chain, so user-registered middleware (rate limit, audit, custom auth) executes inside the span and contributes to the recorded latency. Outbound _meta injection (traceparent/tracestate) sits OUTSIDE any user-registered NotifyInterceptor / RequestInterceptor, so user interceptors observe the same wire form the client will receive — useful for audit logs that want the full envelope.

Inbound trace context resolution order:

  1. params._meta.traceparent / params._meta.tracestate (in-band).
  2. The TraceContext attached to ctx by the transport (e.g., the streamable HTTP transport bridges the HTTP `traceparent` header into ctx per SEP-2028).

In-band wins over out-of-band. Malformed traceparent values are dropped per W3C ("MUST NOT forward"); the span still emits with no parent.

Adapters (P4 ext/otel) MAY update the active TraceContext in ctx after StartSpan so outbound _meta carries the new child span's traceparent. The Noop default leaves ctx unchanged, so outbound _meta carries the inbound traceparent verbatim — still correlates within the trace, just without a finer parent-child link at the next hop. P4 will tighten this for OTel users.

func WithoutListCacheControl added in v0.3.1 

func WithoutListCacheControl() Option

WithoutListCacheControl turns OFF the conformant-by-default SEP-2549 list cache hints (issue 496), so tools/list, prompts/list, resources/list, and resources/templates/list responses omit both ttlMs and cacheScope. Use this only if you deliberately want no cache-control hint on list responses; most servers should keep the default (ttlMs:0, scope:public) or set explicit values with WithListCacheControl.

func WithoutReadResourceCacheControl added in v0.3.1 

func WithoutReadResourceCacheControl() Option

WithoutReadResourceCacheControl turns OFF the conformant-by-default SEP-2549 resources/read cache hints (issue 496), so read responses omit both ttlMs and cacheScope unless a handler sets them per-read. Prefer keeping the default (ttlMs:0, scope:private) or setting explicit values with WithReadResourceCacheControl.

type PendingMap 

type PendingMap = pendingMap

PendingMap is a type alias for SyncMap used to track pending server-to-client requests. Exported because it's referenced by transport implementations.

type Prompt added in v0.1.14 

type Prompt struct {
	core.PromptDef
	Handler core.PromptHandler
}

Prompt bundles a prompt definition with its handler.

type ProtocolFeatures added in v0.3.1 

type ProtocolFeatures struct {
	// RoutingHeaderValidation enforces SEP-2243 §Server Validation: the
	// Mcp-Method header (always) and Mcp-Name header (for name-carrying
	// methods) MUST match the request body, or the server rejects with
	// HTTP 400 + JSON-RPC -32020.
	RoutingHeaderValidation bool

	// StatelessMetaRequired enforces SEP-2575: on the draft line the
	// initialize handshake is gone and every request MUST carry the
	// per-request _meta envelope (protocolVersion / clientInfo /
	// clientCapabilities); missing _meta is rejected with -32602. Servers can
	// opt out of the strict check with WithAllowLegacyOnDraft (see
	// Dispatcher.allowLegacyOnDraft), which is applied by the caller.
	StatelessMetaRequired bool
}

ProtocolFeatures describes the version-gated behaviors active for a given negotiated protocol version. Fields are additive: a zero value means "no version-gated behavior," which is the correct default for every dated release before the draft line.

type QueuedMessage added in v0.2.41 

type QueuedMessage struct {
	Type      QueuedMessageType `json:"type"`
	Timestamp int64             `json:"timestamp"` // Unix milliseconds
	Message   json.RawMessage   `json:"message"`   // JSON-RPC request, notification, response, or error
}

QueuedMessage is a message stored in a task's side-channel queue. During the tasks/result long-poll, the server dequeues these and delivers them to the client (e.g., elicitation or sampling requests).

type QueuedMessageType added in v0.2.41 

type QueuedMessageType string

QueuedMessageType identifies the kind of message in the queue. 

const (
	QueuedMessageRequest      QueuedMessageType = "request"
	QueuedMessageNotification QueuedMessageType = "notification"
	QueuedMessageResponse     QueuedMessageType = "response"
	QueuedMessageError        QueuedMessageType = "error"
)

type Registry 

type Registry struct {
	OnChange RegistryChangeFunc // called after mutations; nil = no notification
	// contains filtered or unexported fields
}

Registry holds all tool, resource, prompt, and completion registrations. Shared by reference across all session Dispatchers so that dynamic adds/removes are immediately visible to every session. Protected by mu: readers (list/call handlers) acquire RLock, writers (AddTool/RemoveTool etc.) acquire Lock.

When OnChange is set, it is called after every mutation with the appropriate list_changed notification method. The Server wires this to Broadcast so that connected clients are notified automatically.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry creates an empty registry with initialized maps.

func (*Registry) AddCompletion 

func (r *Registry) AddCompletion(refType, name string, handler core.CompletionHandler)

AddCompletion adds a completion handler. Thread-safe.

func (*Registry) AddPrompt 

func (r *Registry) AddPrompt(def core.PromptDef, handler core.PromptHandler) error

AddPrompt adds a prompt to the registry. Thread-safe. Broadcasts notifications/prompts/list_changed if OnChange is set.

Compiles any per-argument Schema fields at registration time. Arguments without a Schema bypass validation. Invalid schemas are rejected with a descriptive error (fail-fast).

func (*Registry) AddResource 

func (r *Registry) AddResource(def core.ResourceDef, handler core.ResourceHandler)

AddResource adds a resource to the registry. Thread-safe. Broadcasts notifications/resources/list_changed if OnChange is set.

func (*Registry) AddResourceTemplate 

func (r *Registry) AddResourceTemplate(def core.ResourceTemplate, handler core.TemplateHandler)

AddResourceTemplate adds a resource template to the registry. Thread-safe. Broadcasts notifications/resources/list_changed if OnChange is set.

func (*Registry) AddTool 

func (r *Registry) AddTool(def core.ToolDef, handler core.ToolHandler) error

AddTool adds a tool to the registry. Thread-safe. Broadcasts notifications/tools/list_changed if OnChange is set.

If def.InputSchema is set, it is compiled at registration time. Invalid schemas are rejected with a descriptive error (fail-fast). Compiled schemas are cached on the internal tool entry and used by the dispatcher to validate arguments before handler invocation.

func (*Registry) RemovePrompt 

func (r *Registry) RemovePrompt(name string) bool

RemovePrompt removes a prompt by name. Returns true if it existed. Thread-safe. Broadcasts notifications/prompts/list_changed if the prompt was removed.

func (*Registry) RemoveResource 

func (r *Registry) RemoveResource(uri string) bool

RemoveResource removes a resource by URI. Returns true if it existed. Thread-safe. Broadcasts notifications/resources/list_changed if the resource was removed.

func (*Registry) RemoveResourceTemplate 

func (r *Registry) RemoveResourceTemplate(uriTemplate string) bool

RemoveResourceTemplate removes a resource template by URI template. Returns true if it existed. Thread-safe. Broadcasts notifications/resources/list_changed if the template was removed.

func (*Registry) RemoveTool 

func (r *Registry) RemoveTool(name string) bool

RemoveTool removes a tool by name. Returns true if it existed. Thread-safe. Broadcasts notifications/tools/list_changed if the tool was removed.

func (*Registry) SetToolCallbacks added in v0.2.41 

func (r *Registry) SetToolCallbacks(name string, cb *TaskCallbacks)

SetToolCallbacks associates per-tool task callbacks with a registered tool. The tool must already be registered. Thread-safe.

func (*Registry) ToolCallbacks added in v0.2.41 

func (r *Registry) ToolCallbacks(name string) *TaskCallbacks

ToolCallbacks returns the per-tool task callbacks for a tool, or nil.

func (*Registry) ToolDef added in v0.2.41 

func (r *Registry) ToolDef(name string) (core.ToolDef, bool)

ToolDef returns the definition for a registered tool, or false if not found. Read-only accessor for use by middleware and extensions that need to inspect tool metadata (e.g., Execution.TaskSupport) without access to the handler or compiled schema.

type RegistryChangeFunc 

type RegistryChangeFunc func(method string)

RegistryChangeFunc is called after a registry mutation with the notification method that should be broadcast (e.g., "notifications/tools/list_changed").

type RequestInterceptor added in v0.2.29 

type RequestInterceptor func(ctx context.Context, method string, params any, next core.RequestFunc) (json.RawMessage, error)

RequestInterceptor wraps outgoing server-to-client requests (sampling, elicitation) before they reach the transport. Call next to continue the chain, or return an error to reject the request.

Example — log sampling requests: 

server.WithRequestInterceptor(func(ctx context.Context, method string, params any, next core.RequestFunc) (json.RawMessage, error) {
    log.Printf("→ request %s", method)
    return next(ctx, method, params)
})

type Resource added in v0.1.14 

type Resource struct {
	core.ResourceDef
	Handler core.ResourceHandler
}

Resource bundles a resource definition with its handler.

type ResourceTemplate added in v0.1.14 

type ResourceTemplate struct {
	core.ResourceTemplate
	Handler core.TemplateHandler
}

ResourceTemplate bundles a resource template definition with its handler.

type ResourcesUpdatedReceiver added in v0.2.47 

type ResourcesUpdatedReceiver struct {
	// contains filtered or unexported fields
}

ResourcesUpdatedReceiver is the reference NotificationRelayReceiver for the subscription-shaped notifications/resources/updated. On Receive it extracts the URI from the payload and dispatches via Server.NotifyResourceUpdatedLocal — fires every locally-subscribed session whose subscribe(URI) matches, WITHOUT re-publishing via the NotificationRelay (which would loop).

Each replica's subscriptionRegistry independently filters by its own per-URI subscriber set; clients subscribed via THIS replica's session hear the cross-replica notification, clients on other replicas hear it via THEIR own ResourcesUpdatedReceiver instance.

func NewResourcesUpdatedReceiver added in v0.2.47 

func NewResourcesUpdatedReceiver(srv *Server) *ResourcesUpdatedReceiver

NewResourcesUpdatedReceiver constructs a ResourcesUpdatedReceiver wired to srv. The Server's existing subscriptionRegistry handles per-URI subscriber lookup; this receiver just adapts the cross- replica notification shape into that local call.

func (*ResourcesUpdatedReceiver) Receive added in v0.2.47 

func (r *ResourcesUpdatedReceiver) Receive(_ context.Context, method string, params any)

Receive implements NotificationRelayReceiver. Expects params to be a core.ResourceUpdatedNotification (the type subscriptionRegistry.notify publishes) — also accepts map[string]any with a "uri" field for transports that decoded generically. Unknown shapes are silently dropped.

type SSEData 

type SSEData struct {
	// contains filtered or unexported fields
}

SSEData represents SSE event data that is either raw text or pre-encoded JSON. MCP uses both: the "endpoint" event carries a plain URL string, while "message" events carry JSON-RPC response objects. This type implements json.Marshaler so the servicekit JSONCodec passes the bytes through without double-encoding.

func SSEJSON 

func SSEJSON(j json.RawMessage) SSEData

SSEJSON creates an SSEData containing pre-encoded JSON bytes.

func SSEText 

func SSEText(s string) SSEData

SSEText creates an SSEData containing raw text that will not be JSON-encoded.

type Server 

type Server struct {
	// contains filtered or unexported fields
}

Server is an MCP server that can run over multiple transports.

func NewServer 

func NewServer(info core.ServerInfo, opts ...Option) *Server

NewServer creates an MCP server with the given identity and options.

func (*Server) Broadcast 

func (s *Server) Broadcast(ctx context.Context, method string, params any)

Broadcast sends a JSON-RPC notification to ALL connected sessions across all transports. Unlike NotifyResourceUpdated (which targets only sessions that have called resources/subscribe for a specific URI), Broadcast fans out unconditionally to every session with push capability.

Typical use cases: notifications/tools/list_changed, notifications/prompts/list_changed, or application-level broadcasts.

Safe to call from any goroutine. No-op if no sessions are connected. Sessions without push capability (e.g., Streamable HTTP without GET SSE stream) are silently skipped. Does not hold the server mutex during notification delivery.

Note: only reaches sessions registered through Handler() (SSE and Streamable HTTP transports). In-process transports manage their own notification delivery via WithNotificationHandler.

ctx threads through SEP-414 trace context. When ctx carries a non-zero core.TraceContext, the inbound traceparent / tracestate are injected into the notification params under `_meta` before fan-out, so SSE subscribers see the same trace identity the originating yield ran under. Existing caller-set `_meta.traceparent` on params wins (delegates to core.InjectTraceContextIntoParams's contract).

The injection happens once at the Server level rather than per transport because the per-session notifyFunc the transport dispatchers expose is the BASE notifyFunc — the per-request trace_middleware wrap (`core.WrapSessionNotifyFunc`) targets `sc.notify` on a per-request SessionCtx, which broadcast doesn't run inside. Injecting once at the call site keeps tracing concerns out of the transport-level broadcast loops.

Example: 

// After registering a new tool at runtime:
srv.Broadcast(ctx, "notifications/tools/list_changed", nil)

func (*Server) BroadcastToSessions added in v0.2.47 

func (s *Server) BroadcastToSessions(ctx context.Context, method string, params any)

BroadcastToSessions delivers a notification to every locally-connected client via the per-transport session broadcasters. UNLIKE Broadcast, BroadcastToSessions does NOT invoke the installed NotificationRelay — it's the local-only path Pattern B receivers call after dropping self-publishes, so they don't re-publish through the relay and loop.

External callers should normally use Broadcast (which fires the relay + local). BroadcastToSessions is the entry point for relay receive callbacks (server.CapabilityBroadcastReceiver and equivalent adapters) and tests that want to verify session fan-out independently of the relay path.

Trace context injection: BroadcastToSessions does NOT re-inject _meta.traceparent — the relay receive path is responsible for surfacing the upstream trace context onto ctx, and the params envelope already carries any caller-set _meta. Re-injecting here would double-stamp on the local-fanout-only path called from Broadcast above.

func (*Server) CheckAuth 

func (s *Server) CheckAuth(r *http.Request) (*core.Claims, error)

CheckAuth validates an HTTP request against the server's auth configuration. Returns the authenticated claims (if the validator provides them) and any error. Returns (nil, nil) if no auth is configured.

func (*Server) CloseAllSessions 

func (s *Server) CloseAllSessions()

CloseAllSessions terminates all active sessions across all transports.

func (*Server) CloseSession 

func (s *Server) CloseSession(id string) bool

CloseSession terminates an active session by ID across all transports. Returns true if the session was found and closed.

func (*Server) Dispatch 

func (s *Server) Dispatch(ctx context.Context, req *core.Request) (*core.Response, error)

Dispatch routes a JSON-RPC request through the server's dispatch layer.

Returns the JSON-RPC response and an optional transport-level error. A non-nil error (typically *core.AuthError) indicates middleware short-circuited at the transport layer (e.g., scope step-up); callers should map it to an HTTP-level response via writeAuthError.

func (*Server) HandleMethod added in v0.2.32 

func (s *Server) HandleMethod(method string, h MethodHandler)

HandleMethod registers a handler for a custom JSON-RPC method. Panics if the method is a built-in MCP spec method.

func (*Server) Handler 

func (s *Server) Handler(opts ...TransportOption) http.Handler

Handler returns an http.Handler implementing MCP transports. By default, only the legacy SSE transport is enabled. Use WithStreamableHTTP(true) to enable the Streamable HTTP transport (MCP 2025-03-26). Both transports can be enabled simultaneously for backward compatibility.

func (*Server) IsPublicMethod added in v0.2.31 

func (s *Server) IsPublicMethod(method string) bool

IsPublicMethod returns true if the given JSON-RPC method is in the server's public method set (configured via WithPublicMethods). Public methods bypass auth and are dispatched without claims.

func (*Server) ListenAndServe 

func (s *Server) ListenAndServe(opts ...TransportOption) error

ListenAndServe starts the HTTP transport(s) with graceful shutdown support. On SIGTERM/SIGINT it stops accepting new connections, closes active sessions, drains in-flight requests, and exits.

func (*Server) NotifyResourceUpdated 

func (s *Server) NotifyResourceUpdated(uri string)

NotifyResourceUpdated sends a notifications/resources/updated notification to all clients that have subscribed to the given resource URI. This is the application-facing API for triggering resource change notifications.

Safe to call from any goroutine. No-op if subscriptions are not enabled or no clients are subscribed to the URI.

When a NotificationRelay is installed via WithNotificationRelay, the notification reaches subscribers on every replica — the receive side (typically a server.ResourcesUpdatedReceiver wired into a NotificationRouter) calls notifyLocal on each replica's own subscriptionRegistry.

Example: 

// After updating config.yaml on disk:
srv.NotifyResourceUpdated("file:///data/config.yaml")

func (*Server) NotifyResourceUpdatedLocal added in v0.2.47 

func (s *Server) NotifyResourceUpdatedLocal(uri string)

NotifyResourceUpdatedLocal sends a notifications/resources/updated notification to LOCAL subscribers only — does NOT publish via the installed NotificationRelay. Used by ResourcesUpdatedReceiver.Receive to deliver a cross-replica-received notification without re-publishing.

External callers should normally use NotifyResourceUpdated.

func (*Server) RecordSessionDelta added in v0.2.47 

func (s *Server) RecordSessionDelta(ctx context.Context, delta int64)

RecordSessionDelta lets transports report session-lifecycle deltas to the metrics seam without each transport carrying a MeterProvider of its own. delta is +1 on session-create, -1 on session-expire / DELETE.

No-op when metrics are disabled (s.sessionsActive is nil) — the transport calls this unconditionally and the metrics install state decides whether anything happens. Hot-path cost on the disabled branch is one nil-check.

func (*Server) Register added in v0.1.14 

func (s *Server) Register(items ...any)

Register registers one or more tools, resources, resource templates, or prompts using single-struct registration. Each argument must be a Tool, Resource, ResourceTemplate, or Prompt value.

The existing two-argument methods (RegisterTool, RegisterResource, etc.) remain available for backward compatibility.

Example: 

srv.Register(
    server.Tool{ToolDef: core.ToolDef{Name: "a"}, Handler: handlerA},
    server.Resource{ResourceDef: core.ResourceDef{URI: "test://b"}, Handler: handlerB},
    server.Prompt{PromptDef: core.PromptDef{Name: "c"}, Handler: handlerC},
)

func (*Server) RegisterCompletion 

func (s *Server) RegisterCompletion(refType, name string, handler core.CompletionHandler)

RegisterCompletion registers a completion handler for argument autocompletion. refType is "ref/prompt" or "ref/resource". name is the prompt name or resource URI template.

func (*Server) RegisterExperimentalPrompt 

func (s *Server) RegisterExperimentalPrompt(def core.PromptDef, handler core.PromptHandler)

RegisterExperimentalPrompt registers a prompt marked as experimental via annotations.

func (*Server) RegisterExperimentalResource 

func (s *Server) RegisterExperimentalResource(def core.ResourceDef, handler core.ResourceHandler)

RegisterExperimentalResource registers a resource marked as experimental via annotations.

func (*Server) RegisterExperimentalTool 

func (s *Server) RegisterExperimentalTool(def core.ToolDef, handler core.ToolHandler)

RegisterExperimentalTool registers a tool marked as experimental via annotations.

func (*Server) RegisterExtension added in v0.2.41 

func (s *Server) RegisterExtension(ext core.ExtensionProvider)

RegisterExtension declares a protocol extension at runtime, the way WithExtension does at construction time. Used by RegisterTasks (and other post-construction hookups like ext/auth) so the extension is advertised in the initialize response without forcing callers to thread an Option through. Must be called before accepting connections.

func (*Server) RegisterPrompt 

func (s *Server) RegisterPrompt(def core.PromptDef, handler core.PromptHandler)

RegisterPrompt adds a prompt to the server.

func (*Server) RegisterResource 

func (s *Server) RegisterResource(def core.ResourceDef, handler core.ResourceHandler)

RegisterResource adds a resource to the server.

func (*Server) RegisterResourceTemplate 

func (s *Server) RegisterResourceTemplate(def core.ResourceTemplate, handler core.TemplateHandler)

RegisterResourceTemplate adds a URI template resource to the server.

func (*Server) RegisterTool 

func (s *Server) RegisterTool(def core.ToolDef, handler core.ToolHandler)

RegisterTool adds a tool to the server.

func (*Server) Registry 

func (s *Server) Registry() *Registry

Registry returns the server's shared registry. All session dispatchers share this registry, so changes (AddTool, RemoveTool, etc.) are immediately visible to every session. When the server has active sessions, mutations automatically broadcast the appropriate notifications/*/list_changed notification.

func (*Server) Run 

func (s *Server) Run(addr string, opts ...TransportOption) error

Run is a convenience entry point that starts the server with Streamable HTTP on the given address. It is equivalent to: 

srv.ListenAndServe(mcpkit.WithStreamableHTTP(true))

with the address set via WithListen. For more control over transport options, use ListenAndServe directly.

Example: 

srv := mcpkit.NewServer(mcpkit.ServerInfo{Name: "my-server", Version: "1.0"})
srv.RegisterTool(def, handler)
srv.Run(":8787")

func (*Server) RunIO added in v0.2.27 

func (s *Server) RunIO(ctx context.Context, r io.Reader, w io.Writer, opts ...StdioOption) error

RunIO runs the MCP server over Content-Length framed JSON-RPC on arbitrary io.Reader/io.Writer streams. Blocks until ctx is cancelled or the reader reaches EOF.

This is the generic version of RunStdio — use it for Unix domain sockets, named pipes, SSH tunnels, or any other stream-based transport. RunStdio is equivalent to RunIO(ctx, os.Stdin, os.Stdout).

Example (Unix socket): 

conn, _ := net.Dial("unix", "/tmp/mcp.sock")
srv.RunIO(ctx, conn, conn)

Example (pipe pair for testing): 

sr, cw := io.Pipe()
cr, sw := io.Pipe()
go srv.RunIO(ctx, sr, sw)
client := client.NewClient("", info, client.WithIOTransport(cr, cw))

func (*Server) RunStdio 

func (s *Server) RunStdio(ctx context.Context, opts ...StdioOption) error

RunStdio runs the MCP server over stdio using Content-Length framed JSON-RPC. Blocks until ctx is cancelled or stdin reaches EOF.

This is the primary entry point for editor-spawned MCP servers. Equivalent to RunIO(ctx, os.Stdin, os.Stdout, opts...).

func (*Server) SetTasksCap added in v0.2.41 

func (s *Server) SetTasksCap(cap *core.TasksCap)

SetTasksCap configures the tasks capability advertised during initialize. Must be called before accepting connections. (V1-only: v2 advertises tasks via the io.modelcontextprotocol/tasks extension instead — see RegisterTasks.)

func (*Server) UseMiddleware added in v0.2.41 

func (s *Server) UseMiddleware(mw ...Middleware)

UseMiddleware appends server-side middleware post-construction. Must be called before accepting connections (same constraint as HandleMethod). For construction-time registration, prefer WithMiddleware().

type ServerRequestError added in v0.2.41 

type ServerRequestError struct {
	Code    int
	Message string
}

ServerRequestError is a typed error returned by sendServerRequest when the client responds with a JSON-RPC error. It wraps the error code and message so callers can inspect the code (e.g., to distinguish method-not-found from transport failures).

func (*ServerRequestError) Error added in v0.2.41 

func (e *ServerRequestError) Error() string

type StatelessSubscriptionScopeFunc added in v0.2.47 

type StatelessSubscriptionScopeFunc func(*http.Request) string

StatelessSubscriptionScopeFunc derives a stable key from an inbound HTTP request to identify the calling client for cap / rate-limit bookkeeping. The legacy wire uses sessionID; the stateless wire has no session, so the operator picks the scoping key that makes sense for their deployment (proxy-supplied client header, source IP, auth-subject, etc.).

Implementations MUST be deterministic: two requests from the same client must produce the same string, and two requests from different clients should not collide. An empty string is treated as a single shared scope (all anonymous requests count against one bucket); pass a function that returns a non-empty placeholder if that is not what you want.

type StdioOption 

type StdioOption func(*stdioConfig)

StdioOption configures the stdio transport.

func WithStdioInput 

func WithStdioInput(r io.Reader) StdioOption

WithStdioInput overrides stdin for the stdio transport. Primarily used for testing with pipe pairs.

func WithStdioLogger 

func WithStdioLogger(l *log.Logger) StdioOption

WithStdioLogger sets a logger for debug output on stderr. Debug logging is separate from the MCP protocol — it goes to stderr, never to stdout.

func WithStdioOutput 

func WithStdioOutput(w io.Writer) StdioOption

WithStdioOutput overrides stdout for the stdio transport. Primarily used for testing with pipe pairs.

type SubscriptionRejectFunc added in v0.2.47 

type SubscriptionRejectFunc func(sessionID, uri, reason string)

SubscriptionRejectFunc is invoked when resources/subscribe is refused because the session has hit the configured cap or rate limit. reason is one of "cap_exceeded" or "rate_limited". The hook is called outside the subscription registry's lock; implementations may block but doing so will not delay other requests on the session.

type TaskCallbacks added in v0.2.41 

type TaskCallbacks struct {
	// GetTask overrides the default TaskStore lookup for the v1 tasks/get
	// handler. Return (result, true) to use the override response.
	// Return (_, false) to fall through to the TaskStore.
	GetTask func(ctx core.MethodContext, taskID string) (core.GetTaskResultV1, bool)

	// GetResult overrides the default TaskStore lookup for the v1 tasks/result
	// handler. Return (result, true) to use the override response.
	// Return (_, false) to fall through to the TaskStore.
	GetResult func(ctx core.MethodContext, taskID string) (core.ToolResult, bool)
}

TaskCallbacks provides optional per-tool overrides for task protocol handlers. When a tool registers callbacks, the tasks/get and tasks/result handlers consult them before falling through to the TaskStore.

This enables the "external proxy" pattern where a tool wraps an external job system (e.g., AWS Step Functions, CI/CD pipelines) and the external system is the source of truth for task state — not the in-memory store.

Designed for extensibility: v2 (SEP-2557/SEP-2663) will add OnCancel and OnInputResponse fields without structural changes.

Currently scoped to v1 — the GetTask callback returns the v1 wire shape (GetTaskResultV1). The v2 task runtime does not yet consult these callbacks.

Example: 

srv.Register(server.Tool{
    ToolDef: core.ToolDef{
        Name:      "deploy",
        Execution: &core.ToolExecution{TaskSupport: core.TaskSupportRequired},
    },
    Handler: deployHandler,
    TaskCallbacks: &server.TaskCallbacks{
        GetTask: func(ctx core.MethodContext, taskID string) (core.GetTaskResultV1, bool) {
            // Query external system for task state
            state, err := stepFunctions.DescribeExecution(taskID)
            if err != nil {
                return core.GetTaskResultV1{}, false // fall through to store
            }
            return core.GetTaskResultV1{TaskInfo: toTaskInfo(state)}, true
        },
    },
})

type TaskContext added in v0.2.41 

type TaskContext struct {
	core.ToolContext
	// contains filtered or unexported fields
}

TaskContext is a typed context for tool handlers running as background tasks. It embeds core.ToolContext and adds task-specific methods:

  • TaskID() — the task's unique identifier
  • TaskElicit() — elicitation via the tasks/result side-channel
  • TaskSample() — sampling via the tasks/result side-channel

Tool handlers retrieve this via GetTaskContext(ctx). It is nil for synchronous (non-task) tool invocations.

Usage: 

func myToolHandler(ctx core.ToolContext, input MyInput) (string, error) {
    tc := tasks.GetTaskContext(ctx)
    if tc != nil {
        result, err := tc.TaskElicit(core.ElicitationRequest{...})
    }
    return "done", nil
}

func GetTaskContext added in v0.2.41 

func GetTaskContext(ctx core.ToolContext) *TaskContext

GetTaskContext retrieves the TaskContext from a tool handler's context, binding it to the provided ToolContext for elicitation/sampling. Returns nil if the tool was not invoked as a task.

func (*TaskContext) ProgressToken added in v0.2.41 

func (tc *TaskContext) ProgressToken() any

ProgressToken returns the client's original _meta.progressToken from the tools/call request. Use this instead of TaskID() for EmitProgress so progress notifications correlate with what the client expects. Returns nil if the client didn't send a progressToken.

func (*TaskContext) SetStatus added in v0.2.41 

func (tc *TaskContext) SetStatus(status core.TaskStatus) error

SetStatus transitions the task to a new status and sends a notifications/tasks/status notification (Phase 6).

func (*TaskContext) TaskElicit added in v0.2.41 

func (tc *TaskContext) TaskElicit(req core.ElicitationRequest) (core.ElicitationResult, error)

TaskElicit asks the client for elicitation input from inside a running v1 task. The request is enqueued on a side-channel proxied by the tasks/result long-poll handler.

Status transitions: working → input_required → working.

func (*TaskContext) TaskID added in v0.2.41 

func (tc *TaskContext) TaskID() string

TaskID returns the task's unique identifier.

func (*TaskContext) TaskSample deprecated added in v0.2.41 

func (tc *TaskContext) TaskSample(req core.CreateMessageRequest) (core.CreateMessageResult, error)

TaskSample asks the client for a sampling/createMessage response from inside a running v1 task. See TaskElicit for the side-channel flow.

Status transitions: working → input_required → working.

Deprecated: per SEP-2577, scheduled for removal in v0.4. See docs/SEP_2577_DEPRECATIONS.md.

type TaskMessageQueue added in v0.2.41 

type TaskMessageQueue interface {
	// Enqueue adds a message to the end of the queue for a task.
	// maxSize of 0 means unbounded.
	Enqueue(taskID string, msg QueuedMessage, maxSize int) error

	// Dequeue removes and returns the first message, or false if empty.
	Dequeue(taskID string) (QueuedMessage, bool)

	// DequeueAll removes and returns all messages for a task.
	DequeueAll(taskID string) []QueuedMessage

	// WaitForMessage blocks until a message is available for the task,
	// or the done channel is closed. Returns false if done was closed.
	WaitForMessage(taskID string, done <-chan struct{}) bool

	// Cleanup removes all queues (for shutdown/testing).
	Cleanup()
}

TaskMessageQueue is a per-task FIFO message queue for side-channel communication during async tool execution. When a background task needs to elicit or sample, it enqueues a request here. The tasks/result long-poll dequeues and delivers these messages to the client.

Implementations must be safe for concurrent use.

type TaskStore added in v0.2.41 

type TaskStore interface {
	// Create persists a new task bound to the given session.
	Create(info core.TaskInfo, sessionID string) error

	// Get returns a task by ID, or false if not found or session mismatch.
	Get(taskID, sessionID string) (core.TaskInfo, bool)

	// Update atomically modifies a task via the provided function.
	// Returns an error if the task doesn't exist or session mismatch.
	Update(taskID, sessionID string, fn func(*core.TaskInfo)) error

	// SetResult stores the tool result for a completed task.
	SetResult(taskID, sessionID string, result core.ToolResult) error

	// GetResult returns the stored tool result, or false if not yet available.
	GetResult(taskID, sessionID string) (core.ToolResult, bool)

	// WaitForResult blocks until the task reaches a terminal state, then
	// returns the result. Respects context cancellation.
	WaitForResult(ctx context.Context, taskID, sessionID string) (core.ToolResult, core.TaskInfo, error)

	// List returns tasks for the given session with cursor-based pagination.
	List(cursor string, limit int, sessionID string) ([]core.TaskInfo, string)

	// WaitForUpdate blocks until the task's state changes, or the context
	// is cancelled. Used by the tasks/result long-poll loop.
	WaitForUpdate(ctx context.Context, taskID, sessionID string) error

	// StoreTerminalResult atomically sets the task's result and transitions
	// to a terminal status. Rejects if the task is already terminal.
	// Replaces the separate SetResult + Update pattern.
	//
	// Per MCP spec 2025-11-25 §Tasks: "The result MUST be stored before
	// the status transitions to a terminal state" — this method ensures
	// atomicity so WaitForResult callers always see the result.
	StoreTerminalResult(taskID, sessionID string, status core.TaskStatus, result core.ToolResult, statusMsg string) error

	// Cancel transitions a non-terminal task to cancelled.
	Cancel(taskID, sessionID string) (core.TaskInfo, error)

	// Cleanup removes all tasks and stops any background timers.
	Cleanup()
}

TaskStore is the interface for task state persistence. Implementations must be safe for concurrent use.

All methods accept a sessionID parameter for session isolation. Empty sessionID means no session binding (backward compatible). When both the task and the caller have a sessionID, they must match.

type TasksConfigV1 added in v0.2.41 

type TasksConfigV1 struct {
	// Store is the task state backend. If nil, an InMemoryTaskStore is used.
	Store TaskStore

	// MessageQueue is the per-task FIFO queue for side-channel messages
	// (elicitation/sampling requests delivered via tasks/result long-poll).
	// If nil, an InMemoryMessageQueue is used.
	MessageQueue TaskMessageQueue

	// Server is the MCP server to register tasks on.
	Server *Server

	// DefaultTTLMs is the default task TTL in milliseconds. Tasks are
	// cleaned up after this duration. Default: 300000 (5 minutes).
	DefaultTTLMs int

	// DefaultPollMs is the suggested poll interval in milliseconds,
	// returned to clients in CreateTaskResult. Default: 1000 (1 second).
	DefaultPollMs int

	// MaxQueueSize is the maximum number of messages per task queue.
	// 0 means unbounded.
	MaxQueueSize int
}

Config holds the options for registering tasks support on an MCP server.

type Tool added in v0.1.14 

type Tool struct {
	core.ToolDef
	Handler core.ToolHandler
	// TaskCallbacks provides optional per-tool overrides for tasks/get and
	// tasks/result handlers. When set, the task protocol consults these
	// callbacks before falling through to the TaskStore. See [TaskCallbacks].
	TaskCallbacks *TaskCallbacks
}

Tool bundles a tool definition with its handler for single-struct registration via Server.Register. This is the recommended way to register tools — it keeps the definition and handler together as a single value, making it easy to build tool registries or load tools from configuration.

Example: 

srv.Register(server.Tool{
    ToolDef: core.ToolDef{Name: "echo", Description: "Echo input", InputSchema: schema},
    Handler: func(ctx core.ToolContext, req core.ToolRequest) (core.ToolResponse, error) {
        return core.TextResult("echoed"), nil
    },
})

func ToolExec added in v0.1.16 

func ToolExec(cfg ExecConfig) Tool

ToolExec creates a Tool that wraps a CLI command execution. The returned Tool can be registered via srv.Register(server.ToolExec(cfg)).

On success (exit code 0), returns TextResult with the combined stdout/stderr. On failure (non-zero exit), returns ErrorResult with the output and exit code. Context cancellation (including Timeout) is propagated to the subprocess.

type TransportOption 

type TransportOption func(*transportConfig)

TransportOption configures the HTTP transports.

func WithAllowedOrigins 

func WithAllowedOrigins(origins ...string) TransportOption

WithAllowedOrigins sets the allowed Origin header values for DNS rebinding protection. When empty (default), only localhost origins are accepted.

func WithEventStore added in v0.1.11 

func WithEventStore(store gohttp.EventStore) TransportOption

WithEventStore sets an optional EventStore for SSE event persistence. When configured, all SSE events (GET SSE stream notifications and POST SSE response events) are stored with unique IDs. Clients that reconnect with a Last-Event-ID header receive missed events via replay.

Pass nil to disable (default). Use gohttp.NewMemoryEventStore(maxPerStream) for an in-memory implementation.

func WithHandlerWrap added in v0.2.44 

func WithHandlerWrap(wrap func(http.Handler) http.Handler) TransportOption

WithHandlerWrap wraps the final HTTP handler (after WithMux composition, if any) before it's served. Useful for cross-cutting concerns that should apply to every route the server exposes — CORS for browser-based MCP hosts, rate limiting, request tracing, etc.

Multiple WithHandlerWrap options compose: the first one registered is innermost (closest to the MCP handler), and the last one registered is outermost (the first to see an incoming request and last to see the outgoing response). This matches conventional HTTP middleware ordering.

Example — applying CORS so MCPJam (browser-based) can connect: 

cors := middleware.CORS(nil,
    middleware.CORSAllowMethods("GET", "POST", "DELETE", "OPTIONS"),
    middleware.CORSAllowHeaders("Content-Type", "Authorization", "Mcp-Session-Id"),
    middleware.CORSExposeHeaders("Mcp-Session-Id"),
)
srv.ListenAndServe(
    server.WithStreamableHTTP(true),
    server.WithMux(func(m *http.ServeMux) { m.HandleFunc("/approve", ...) }),
    server.WithHandlerWrap(cors),
)

func WithKeepalive added in v0.1.11 

func WithKeepalive(interval time.Duration, maxFailures int) TransportOption

WithKeepalive enables application-level keepalive pings for session liveness detection. When enabled, the server periodically sends JSON-RPC ping requests to clients via their GET SSE stream. If maxFailures consecutive pings fail (timeout or error), the session is expired.

interval controls how often pings are sent (e.g., 30*time.Second). maxFailures is the threshold before session cleanup (e.g., 3). Set interval to 0 to disable (default).

func WithKeepalivePeriod 

func WithKeepalivePeriod(d time.Duration) TransportOption

WithKeepalivePeriod sets the interval for SSE keepalive comments.

func WithMaxSessions 

func WithMaxSessions(n int) TransportOption

WithMaxSessions limits the number of concurrent sessions.

func WithMux added in v0.2.41 

func WithMux(setup func(*http.ServeMux)) TransportOption

WithMux registers additional HTTP routes on the server's mux alongside the MCP transport handler. Use this to mount auth discovery (PRM), health checks, or other endpoints without dropping out of srv.Run()'s graceful shutdown.

Example: 

srv.Run(":8080",
    server.WithStreamableHTTP(true),
    server.WithMux(func(mux *http.ServeMux) {
        auth.MountAuth(mux, authCfg)
        mux.HandleFunc("/healthz", healthHandler)
    }),
)

func WithPrefix 

func WithPrefix(p string) TransportOption

WithPrefix sets the URL path prefix for transport endpoints.

func WithPublicURL 

func WithPublicURL(u string) TransportOption

WithPublicURL sets the public base URL used in the SSE endpoint event. Use this when the server is behind a reverse proxy.

func WithSSE 

func WithSSE(enabled bool) TransportOption

WithSSE enables or disables the legacy SSE transport (MCP 2024-11-05).

func WithSSEGracePeriod added in v0.1.19 

func WithSSEGracePeriod(d time.Duration) TransportOption

WithSSEGracePeriod sets the grace period for SSE sessions after disconnect. When an SSE connection closes, the session stays alive for this duration. If the client reconnects with the same session ID (via ?sessionId= query param), it resumes the existing session and replays missed events via Last-Event-ID. Requires WithEventStore for event replay.

Default is 0 (no grace period — sessions die immediately on disconnect, backward compatible with pre-v0.1.17 behavior).

Security: reconnection requires the same auth principal as the original session. Session IDs are cryptographically random.

func WithSessionTimeout 

func WithSessionTimeout(d time.Duration) TransportOption

WithSessionTimeout sets the idle timeout for Streamable HTTP sessions. Sessions that receive no POST requests for this duration are automatically expired and cleaned up. Active requests (in-flight tool calls, open GET SSE streams) pause the timer so sessions are never closed mid-execution. Default is 0 (no timeout — sessions persist until explicit DELETE or server restart).

func WithStateless 

func WithStateless(enabled bool) TransportOption

WithStateless enables stateless mode for the Streamable HTTP transport. In stateless mode, every request gets a fresh Dispatcher — no session storage, no Mcp-Session-Id header, no state carried across requests. The initialize handshake is auto-performed per request.

Use for simple tool servers that don't need session state (e.g., single-tool APIs, serverless functions, CLI wrappers).

func WithStatelessMode added in v0.2.47 

func WithStatelessMode(m stateless.Mode) TransportOption

WithStatelessMode pins the SEP-2575 wire mode for this server. The full enum, parser, env-var resolver, and default lifecycle live in server/stateless/mode.go — this file only carries the TransportOption wrapper so the option API stays in package server alongside every other With* TransportOption.

Precedence (high → low):

  1. server.WithStatelessMode(...) ← this option
  2. MCPKIT_STATELESS_MODE env var ← seeded by stateless.ResolveMode
  3. stateless.DefaultMode ← package var, mutable in init()

See the stateless.Mode godoc for the full mode table and the orthogonal relationship to WithStateless (process-architecture).

func WithStreamableHTTP 

func WithStreamableHTTP(enabled bool) TransportOption

WithStreamableHTTP enables or disables the Streamable HTTP transport (MCP 2025-03-26).

type ValidationError added in v0.1.22 

type ValidationError struct {
	Path    string `json:"path"`
	Keyword string `json:"keyword"`
	Message string `json:"message"`
}

ValidationError is one violation reported by schema validation. Path is a JSON Pointer to the offending value (e.g., "/age"), Keyword is the JSON Schema keyword that failed (e.g., "type", "format"), and Message is a human-readable description.

type ValidationErrors added in v0.1.22 

type ValidationErrors struct {
	Errors []ValidationError `json:"errors"`
}

ValidationErrors is the payload returned in the error.data field of JSON-RPC -32602 responses when argument validation fails.

Source Files

View all Source files

Directories

Path Synopsis
Package stateless implements the SEP-2575 stateless wire — server/discover, per-request _meta envelope dispatch, subscriptions/listen, and the new HTTP-status error mapping.
 Package stateless implements the SEP-2575 stateless wire — server/discover, per-request _meta envelope dispatch, subscriptions/listen, and the new HTTP-status error mapping.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.