README
¶
server
MCP server implementation: Dispatcher, transports, middleware, subscriptions.
What belongs here
Serverstruct 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.Transportimplementation 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) — callsRefValidatoron 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
¶
Index ¶
- Constants
- type Dispatcher
- func (d *Dispatcher) Close()
- func (d *Dispatcher) Dispatch(ctx context.Context, req *core.Request) *core.Response
- func (d *Dispatcher) NegotiatedVersion() string
- func (d *Dispatcher) RegisterCompletion(refType, name string, handler core.CompletionHandler)
- func (d *Dispatcher) RegisterPrompt(def core.PromptDef, handler core.PromptHandler)
- func (d *Dispatcher) RegisterResource(def core.ResourceDef, handler core.ResourceHandler)
- func (d *Dispatcher) RegisterResourceTemplate(def core.ResourceTemplate, handler core.TemplateHandler)
- func (d *Dispatcher) RegisterTool(def core.ToolDef, handler core.ToolHandler)
- func (d *Dispatcher) RouteResponse(resp *core.Response) bool
- func (d *Dispatcher) SetNotifyFunc(fn core.NotifyFunc)
- type InProcessOption
- type InProcessTransport
- func (t *InProcessTransport) Call(ctx context.Context, req *core.Request) (*core.Response, error)
- func (t *InProcessTransport) Close() error
- func (t *InProcessTransport) Connect(ctx context.Context) error
- func (t *InProcessTransport) Notify(ctx context.Context, req *core.Request) error
- func (t *InProcessTransport) SessionID() string
- type Middleware
- type MiddlewareFunc
- type Option
- func WithAllowedRoots(roots ...string) Option
- func WithAuth(v core.AuthValidator) Option
- func WithBearerToken(token string) Option
- func WithExtension(ext core.ExtensionProvider) Option
- func WithListen(addr string) Option
- func WithMiddleware(mw ...Middleware) Option
- func WithRequestLogging(logger *log.Logger) Option
- func WithSubscriptions() Option
- func WithToolTimeout(d time.Duration) Option
- type PendingMap
- type SSEData
- type Server
- func (s *Server) Broadcast(method string, params any)
- func (s *Server) CheckAuth(r *http.Request) (*core.Claims, error)
- func (s *Server) CloseAllSessions()
- func (s *Server) CloseSession(id string) bool
- func (s *Server) Dispatch(ctx context.Context, req *core.Request) *core.Response
- func (s *Server) Handler(opts ...TransportOption) http.Handler
- func (s *Server) ListenAndServe(opts ...TransportOption) error
- func (s *Server) NotifyResourceUpdated(uri string)
- func (s *Server) RegisterCompletion(refType, name string, handler core.CompletionHandler)
- func (s *Server) RegisterExperimentalPrompt(def core.PromptDef, handler core.PromptHandler)
- func (s *Server) RegisterExperimentalResource(def core.ResourceDef, handler core.ResourceHandler)
- func (s *Server) RegisterExperimentalTool(def core.ToolDef, handler core.ToolHandler)
- func (s *Server) RegisterPrompt(def core.PromptDef, handler core.PromptHandler)
- func (s *Server) RegisterResource(def core.ResourceDef, handler core.ResourceHandler)
- func (s *Server) RegisterResourceTemplate(def core.ResourceTemplate, handler core.TemplateHandler)
- func (s *Server) RegisterTool(def core.ToolDef, handler core.ToolHandler)
- func (s *Server) Run(addr string, opts ...TransportOption) error
- func (s *Server) RunStdio(ctx context.Context, opts ...StdioOption) error
- type StdioOption
- type TransportOption
- func WithAllowedOrigins(origins ...string) TransportOption
- func WithKeepalivePeriod(d time.Duration) TransportOption
- func WithMaxSessions(n int) TransportOption
- func WithPrefix(p string) TransportOption
- func WithPublicURL(u string) TransportOption
- func WithSSE(enabled bool) TransportOption
- func WithStateless(enabled bool) TransportOption
- func WithStreamableHTTP(enabled bool) TransportOption
Constants ¶
const ErrCodeCancelled = -32800
ErrCodeCancelled is the JSON-RPC error code for a cancelled request.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Dispatcher ¶
type Dispatcher struct {
// 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 ¶
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.
func (*Dispatcher) RegisterResource ¶
func (d *Dispatcher) RegisterResource(def core.ResourceDef, handler core.ResourceHandler)
RegisterResource adds a resource to the dispatcher.
func (*Dispatcher) RegisterResourceTemplate ¶
func (d *Dispatcher) RegisterResourceTemplate(def core.ResourceTemplate, handler core.TemplateHandler)
RegisterResourceTemplate adds a URI template resource to the dispatcher.
func (*Dispatcher) RegisterTool ¶
func (d *Dispatcher) RegisterTool(def core.ToolDef, handler core.ToolHandler)
RegisterTool adds a tool to the dispatcher.
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.
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 ¶
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 ¶
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 Middleware ¶
Middleware intercepts a JSON-RPC request. Call next to continue the chain, or return a *core.Response directly to short-circuit (e.g., reject a request).
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 {
if !limiter.Allow() {
return mcpkit.NewErrorResponse(req.ID, -32000, "rate limit exceeded")
}
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]
type MiddlewareFunc ¶
MiddlewareFunc is the signature for the next handler in the middleware chain.
type Option ¶
type Option func(*serverOptions)
Option configures a Server.
func WithAllowedRoots ¶
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 ¶
WithBearerToken sets a static bearer token for authentication. Uses constant-time comparison to prevent timing attacks.
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 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 WithRequestLogging ¶
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 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 ¶
WithToolTimeout sets the maximum duration for tool execution.
type PendingMap ¶
PendingMap is a typed alias for sync.Map used to track pending server-to-client requests.
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.
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 ¶
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 ¶
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 ¶
CloseSession terminates an active session by ID across all transports. Returns true if the session was found and closed.
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) 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 ¶
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) 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) 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) 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) 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 the input stream reaches EOF.
This is the primary entry point for editor-spawned MCP servers. The server reads JSON-RPC messages from stdin (or the configured input), dispatches them through the standard Server dispatch pipeline, and writes responses to stdout (or the configured output).
Example:
srv := server.NewServer(core.ServerInfo{Name: "my-server", Version: "1.0"})
srv.RegisterTool(def, handler)
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
defer cancel()
if err := srv.RunStdio(ctx); err != nil {
log.Fatal(err)
}
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 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 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 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 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).
Source Files