server

package
v0.2.45 Latest Latest
Warning

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

 Go to latest Published: May 7, 2026 License: Apache-2.0 Imports: 25 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

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 ErrCodeCancelled = -32800

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

Variables

This section is empty.

Functions

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 RegisterTasks added in v0.2.41 

func RegisterTasks(cfg TasksConfig)

RegisterTasks hooks up v2 tasks support on the given server:

  • Advertises the io.modelcontextprotocol/tasks extension in initialize (replacing the v1 ServerCapabilities.Tasks declaration).
  • Installs middleware that intercepts tools/call for task-eligible tools (server-directed, no client task param needed). Task creation is gated on the client supporting the extension — either at session level (initialize handshake) or per-request (SEP-2575 _meta).
  • Registers tasks/get and tasks/cancel handlers, gated on session-level extension support; otherwise the handlers return -32601 (method not found) so unsupported clients don't see a tasks surface they didn't ask for.
  • Does NOT register tasks/result or tasks/list (removed in v2).
  • Does NOT call SetTasksCap — v2 tasks live under capabilities.extensions, not the v1 ServerCapabilities.Tasks slot.

Must be called before accepting connections.

func RegisterTasksHybrid added in v0.2.41 

func RegisterTasksHybrid(cfg TasksHybridConfig)

RegisterTasksHybrid hooks up both v1 and v2 task surfaces on the given server. Capability advertisement is dual:

  • capabilities.tasks (v1 ServerCapabilities.Tasks) is set so v1 clients can negotiate the legacy path.
  • capabilities.extensions[io.modelcontextprotocol/tasks] (v2) is set so v2 clients can negotiate the extension path.

Per-request routing:

  • tools/call: both middlewares run (v2 first; v2 only fires if the client negotiated the extension AND the tool is task-eligible; v1 only fires if the client sent a `task` hint AND the tool supports it). A v2-aware client gets v2 task creation; a v1 client gets v1. A client that negotiated both AND sent a task hint gets v2 (modern wins).
  • tasks/get / tasks/cancel: dispatched to the v2 handler when the client negotiated the extension; otherwise the v1 handler.
  • tasks/update: v2 only — gated on extension negotiation.
  • tasks/result / tasks/list: v1 only — gated on the v1 capability declaration so v2-only clients don't see them.

Must be called before accepting connections. Calling RegisterTasksHybrid in addition to RegisterTasks or RegisterTasksV1 on the same server is undefined — the LAST HandleMethod registration wins.

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 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) *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 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 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 WithAllowedRoots 

func WithAllowedRoots(roots ...string) Option

WithAllowedRoots restricts tool cwd to the given directory prefixes.

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 WithListTTL added in v0.2.42 

func WithListTTL(seconds int) Option

WithListTTL configures the SEP-2549 cache-freshness hint (in seconds) attached to every tools/list, prompts/list, resources/list, and resources/templates/list response. The TTL 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 `"ttl": 0` meaning "do not cache" — clients SHOULD re-fetch every time the list is needed.
  • Positive values send `"ttl": N` meaning "fresh for N seconds"; clients SHOULD NOT re-fetch before TTL expires unless they receive a list_changed notification.

The hint applies uniformly to all four list endpoints. Per-endpoint granularity (e.g. shorter TTL for tools that change often, longer for prompts) is a future option if the simple uniform default proves too coarse.

func WithListen 

func WithListen(addr string) Option

WithListen sets the HTTP listen address.

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 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 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 IncompleteResult 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 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.

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 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.

func WithToolTimeout 

func WithToolTimeout(d time.Duration) Option

WithToolTimeout sets the maximum duration for tool execution.

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 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 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(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.

Example: 

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

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.

Example: 

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

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 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 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 task.

Behavior depends on which task runtime owns this TaskContext:

  • v2 (SEP-2663): the request is stashed on the task's inputState under a monotonic key, the task transitions to input_required, and the goroutine blocks on a per-key waiter channel. The client observes the pending request via tasks/get's DetailedTask.InputRequests and resumes the task by sending the matching response via tasks/update. ctx cancellation unblocks the wait with the corresponding context error.
  • v1 (legacy): the request is enqueued on a side-channel proxied by the tasks/result long-poll handler.

Status transitions for both paths: 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 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 task. See TaskElicit for the v1 vs v2 routing.

Status transitions: working → input_required → working.

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 TasksConfig added in v0.2.41 

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

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

	// DefaultTTLMs is the default task TTL in integer milliseconds. Per
	// SEP-2663 the wire surfaces ttlMs and the store also uses ms, so this
	// value flows through unchanged. Default: 300000 (5 minutes).
	DefaultTTLMs int

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

	// RequestStateKey is the HMAC-SHA256 key the server uses to sign and
	// verify SEP-2322 requestState tokens. When non-nil, requestState is
	// returned as `<base64url-hmac>.<base64url-payload>` where the payload
	// is JSON {"taskId":"...", "exp":<unix-seconds>} — clients echo it
	// verbatim and the server rejects tampered or expired tokens with
	// -32602 on tasks/get / tasks/update / tasks/cancel.
	//
	// SEP-2663 says servers MUST treat requestState as attacker-controlled.
	// Production deployments SHOULD set this. nil = legacy plaintext mode
	// (requestState == taskID), kept for backward compat with existing
	// tests and minimal-config setups.
	RequestStateKey []byte

	// RequestStateTTL is how long a signed requestState stays valid. When
	// 0, defaults to 24h. Has no effect when RequestStateKey is nil.
	RequestStateTTL time.Duration
}

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

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 TasksHybridConfig added in v0.2.41 

type TasksHybridConfig struct {
	// Server is the MCP server to register tasks on. Required; the embedded
	// V1.Server / V2.Server fields are overridden so callers don't have to
	// thread the same pointer twice.
	Server *Server

	// V1 carries the v1-side configuration (TaskStore, TaskMessageQueue,
	// DefaultTTLMs, DefaultPollMs, MaxQueueSize). All optional — defaults
	// match RegisterTasksV1's standalone behavior.
	V1 TasksConfigV1

	// V2 carries the v2-side configuration (TaskStore, DefaultTTLMs,
	// DefaultPollMs). All optional — defaults match RegisterTasks's
	// standalone behavior.
	V2 TasksConfig
}

TasksHybridConfig wires both the v1 (RegisterTasksV1) and v2 (RegisterTasks) task surfaces onto a single server, with per-request dispatch based on which capability the client negotiated during initialize. Use this when you need to keep older v1 clients working while accepting newer v2 clients on the same endpoint — e.g., during a rolling-upgrade window.

If your server only needs to support one path, prefer the dedicated RegisterTasks (v2) or RegisterTasksV1 (v1) entry points — they're simpler and have no dispatching cost per request.

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.ToolResult, 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 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

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.