pool

package
v0.9.1 Latest Latest
Warning

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

 Go to latest Published: Aug 22, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package pool provides stdio MCP server subprocess pooling and management. It handles concurrent request processing, health monitoring, and lifecycle management for multiple stdio-based MCP servers.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type HTTPClientPool added in v0.5.0 

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

func NewHTTPClientPool added in v0.5.0 

func NewHTTPClientPool(logger *slog.Logger) *HTTPClientPool

func (*HTTPClientPool) Close added in v0.5.0 

func (p *HTTPClientPool) Close() error

func (*HTTPClientPool) GetServerState added in v0.5.0 

func (p *HTTPClientPool) GetServerState(name string) (ServerState, error)

func (*HTTPClientPool) HasServer added in v0.5.0 

func (p *HTTPClientPool) HasServer(name string) bool

func (*HTTPClientPool) IsServerMCPInitialized added in v0.7.1 

func (p *HTTPClientPool) IsServerMCPInitialized(name string) bool

func (*HTTPClientPool) ListServers added in v0.5.0 

func (p *HTTPClientPool) ListServers() []string

func (*HTTPClientPool) MarkServerMCPInitialized added in v0.7.1 

func (p *HTTPClientPool) MarkServerMCPInitialized(name string)

func (*HTTPClientPool) RestartServer added in v0.5.0 

func (p *HTTPClientPool) RestartServer(ctx context.Context, name string) error

func (*HTTPClientPool) SendRequest added in v0.5.0 

func (p *HTTPClientPool) SendRequest(ctx context.Context, serverName string, req *proxy.JSONRPCRequest, timeout time.Duration) (*proxy.JSONRPCResponse, error)

func (*HTTPClientPool) SendRequestToServer added in v0.5.0 

func (p *HTTPClientPool) SendRequestToServer(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration) (*Response, error)

func (*HTTPClientPool) SendRequestToServerWithID added in v0.5.0 

func (p *HTTPClientPool) SendRequestToServerWithID(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration, id int) (*Response, error)

func (*HTTPClientPool) SendServerNotification added in v0.5.0 

func (p *HTTPClientPool) SendServerNotification(ctx context.Context, name string, method string, params map[string]interface{}) error

func (*HTTPClientPool) ServerCount added in v0.5.0 

func (p *HTTPClientPool) ServerCount() int

func (*HTTPClientPool) StartServer added in v0.5.0 

func (p *HTTPClientPool) StartServer(ctx context.Context, config *migrate.ServerConfig) error

type HTTPClientServer added in v0.5.0 

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

func NewHTTPClientServer added in v0.5.0 

func NewHTTPClientServer(name string, config *migrate.ServerConfig, logger *slog.Logger) *HTTPClientServer

func (*HTTPClientServer) CallTool added in v0.5.0 

func (s *HTTPClientServer) CallTool(ctx context.Context, name string, args map[string]interface{}) (*mcp.CallToolResult, error)

func (*HTTPClientServer) Close added in v0.5.0 

func (s *HTTPClientServer) Close() error

func (*HTTPClientServer) Initialize added in v0.5.0 

func (s *HTTPClientServer) Initialize(ctx context.Context) error

func (*HTTPClientServer) ListTools added in v0.5.0 

func (s *HTTPClientServer) ListTools(ctx context.Context) ([]mcp.Tool, error)

type HealthCheckResult 

type HealthCheckResult struct {
	ServerName string
	Status     HealthStatus
	LatencyMs  float64
	Error      string
	CheckedAt  time.Time
}

type HealthChecker 

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

func NewHealthChecker 

func NewHealthChecker(pool *StdioPool, logger *slog.Logger) *HealthChecker

func (*HealthChecker) CheckServer 

func (hc *HealthChecker) CheckServer(ctx context.Context, name string) HealthCheckResult

func (*HealthChecker) GetAllHealth 

func (hc *HealthChecker) GetAllHealth() map[string]HealthCheckResult

func (*HealthChecker) GetServerHealth 

func (hc *HealthChecker) GetServerHealth(name string) (HealthStatus, error)

func (*HealthChecker) RegisterServer 

func (hc *HealthChecker) RegisterServer(name string)

func (*HealthChecker) SetMaxFailures added in v0.9.0 

func (hc *HealthChecker) SetMaxFailures(n int)

func (*HealthChecker) Start 

func (hc *HealthChecker) Start(ctx context.Context, interval time.Duration)

func (*HealthChecker) Stop 

func (hc *HealthChecker) Stop()

Stop signals the checker loop to exit. It is idempotent.

func (*HealthChecker) UnregisterServer 

func (hc *HealthChecker) UnregisterServer(name string)

type HealthStatus 

type HealthStatus string
const (
	HealthUnknown   HealthStatus = "unknown"
	HealthHealthy   HealthStatus = "healthy"
	HealthDegraded  HealthStatus = "degraded"
	HealthUnhealthy HealthStatus = "unhealthy"
	HealthError     HealthStatus = "error"
)

type PingRequest 

type PingRequest struct {
	ID      interface{} `json:"id"`
	JSONRPC string      `json:"jsonrpc"`
	Method  string      `json:"method"`
}

type PingResponse 

type PingResponse struct {
	ID      interface{}          `json:"id"`
	JSONRPC string               `json:"jsonrpc"`
	Result  json.RawMessage      `json:"result,omitempty"`
	Error   *errors.JSONRPCError `json:"error,omitempty"`
}

type PoolQueueManager 

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

func NewPoolQueueManager 

func NewPoolQueueManager(logger *slog.Logger) *PoolQueueManager

func (*PoolQueueManager) GetOrCreateQueue 

func (qm *PoolQueueManager) GetOrCreateQueue(name string, maxConcurrent int, queueTimeout time.Duration) *ServerQueue

func (*PoolQueueManager) GetQueueStats 

func (qm *PoolQueueManager) GetQueueStats(name string) (active int, pending int, atCapacity bool)

func (*PoolQueueManager) ListQueues 

func (qm *PoolQueueManager) ListQueues() []string

func (*PoolQueueManager) RemoveQueue 

func (qm *PoolQueueManager) RemoveQueue(name string)

type ReconnectSettings added in v0.9.0 

type ReconnectSettings struct {
	// Disabled is the reconnect.enabled=false master switch: crashed servers
	// are left in the error state instead of being auto-restarted. Explicit
	// restarts (request- or operator-triggered) still work.
	Disabled           bool
	MaxRestartAttempts int
	RestartBackoff     time.Duration
	StableWindow       time.Duration
}

ReconnectSettings controls the automatic restart behavior of stdio servers.

type Request 

type Request struct {
	Method   string          `json:"method"`
	Params   json.RawMessage `json:"params,omitempty"`
	ID       interface{}     `json:"id"`
	Timeout  time.Duration   `json:"-"`
	ResultCh chan *Response  `json:"-"`
	ErrorCh  chan error      `json:"-"`
}

Request represents a JSON-RPC request for an MCP server.

func (Request) MarshalJSON added in v0.3.0 

func (r Request) MarshalJSON() ([]byte, error)

MarshalJSON serializes the Request to JSON with JSON-RPC 2.0 formatting.

type RequestQueue 

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

func NewRequestQueue 

func NewRequestQueue(maxSize int, timeout time.Duration, logger *slog.Logger) *RequestQueue

func (*RequestQueue) Dequeue 

func (q *RequestQueue) Dequeue(ctx context.Context) (Request, bool)

func (*RequestQueue) Enqueue 

func (q *RequestQueue) Enqueue(req Request) bool

func (*RequestQueue) IsEmpty 

func (q *RequestQueue) IsEmpty() bool

func (*RequestQueue) IsFull 

func (q *RequestQueue) IsFull() bool

func (*RequestQueue) Size 

func (q *RequestQueue) Size() int

type Response 

type Response struct {
	Result json.RawMessage      `json:"result,omitempty"`
	Error  *errors.JSONRPCError `json:"error,omitempty"`
	ID     interface{}          `json:"id"`
}

Response represents a JSON-RPC response from an MCP server.

type SSEPool added in v0.5.0 

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

func NewSSEPool added in v0.5.0 

func NewSSEPool(logger *slog.Logger) *SSEPool

func (*SSEPool) Close added in v0.5.0 

func (p *SSEPool) Close() error

func (*SSEPool) GetServerState added in v0.5.0 

func (p *SSEPool) GetServerState(name string) (ServerState, error)

func (*SSEPool) HasServer added in v0.5.0 

func (p *SSEPool) HasServer(name string) bool

func (*SSEPool) IsServerMCPInitialized added in v0.7.1 

func (p *SSEPool) IsServerMCPInitialized(name string) bool

func (*SSEPool) ListServers added in v0.5.0 

func (p *SSEPool) ListServers() []string

func (*SSEPool) MarkServerMCPInitialized added in v0.7.1 

func (p *SSEPool) MarkServerMCPInitialized(name string)

func (*SSEPool) RestartServer added in v0.5.0 

func (p *SSEPool) RestartServer(ctx context.Context, name string) error

func (*SSEPool) SendRequest added in v0.5.0 

func (p *SSEPool) SendRequest(ctx context.Context, serverName string, req *proxy.JSONRPCRequest, timeout time.Duration) (*proxy.JSONRPCResponse, error)

func (*SSEPool) SendRequestToServer added in v0.5.0 

func (p *SSEPool) SendRequestToServer(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration) (*Response, error)

func (*SSEPool) SendRequestToServerWithID added in v0.5.0 

func (p *SSEPool) SendRequestToServerWithID(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration, id int) (*Response, error)

func (*SSEPool) SendServerNotification added in v0.5.0 

func (p *SSEPool) SendServerNotification(ctx context.Context, name string, method string, params map[string]interface{}) error

func (*SSEPool) ServerCount added in v0.5.0 

func (p *SSEPool) ServerCount() int

func (*SSEPool) StartServer added in v0.5.0 

func (p *SSEPool) StartServer(ctx context.Context, config *migrate.ServerConfig) error

type SSEServer added in v0.5.0 

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

func NewSSEServer added in v0.5.0 

func NewSSEServer(name string, config *migrate.ServerConfig, logger *slog.Logger) *SSEServer

func (*SSEServer) CallTool added in v0.5.0 

func (s *SSEServer) CallTool(ctx context.Context, name string, args map[string]interface{}) (*mcp.CallToolResult, error)

func (*SSEServer) Close added in v0.5.0 

func (s *SSEServer) Close() error

func (*SSEServer) Initialize added in v0.5.0 

func (s *SSEServer) Initialize(ctx context.Context) error

func (*SSEServer) ListTools added in v0.5.0 

func (s *SSEServer) ListTools(ctx context.Context) ([]mcp.Tool, error)

type ServerHandle 

type ServerHandle struct {
	Name  string
	State ServerState
	Stats ServerStats
}

type ServerQueue 

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

func NewServerQueue 

func NewServerQueue(name string, maxConcurrent int, queueTimeout time.Duration, logger *slog.Logger) *ServerQueue

func (*ServerQueue) Acquire 

func (sq *ServerQueue) Acquire(timeout time.Duration) bool

func (*ServerQueue) Dequeue 

func (sq *ServerQueue) Dequeue(ctx context.Context) (Request, bool)

func (*ServerQueue) Enqueue 

func (sq *ServerQueue) Enqueue(req Request) bool

func (*ServerQueue) IsAtCapacity 

func (sq *ServerQueue) IsAtCapacity() bool

func (*ServerQueue) PendingCount 

func (sq *ServerQueue) PendingCount() int

func (*ServerQueue) Release 

func (sq *ServerQueue) Release()

type ServerSource added in v0.5.0 

type ServerSource interface {
	SendRequestToServer(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration) (*Response, error)
	SendRequestToServerWithID(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration, id int) (*Response, error)
	SendServerNotification(ctx context.Context, name string, method string, params map[string]interface{}) error
	ListServers() []string
	GetServerState(name string) (ServerState, error)
	RestartServer(ctx context.Context, name string) error
	IsServerMCPInitialized(name string) bool
	MarkServerMCPInitialized(name string)
	Close() error
}

ServerSource is the interface for sending requests to MCP servers.

type ServerState 

type ServerState string

ServerState represents the current state of an MCP server. 

const (
	StateIdle         ServerState = "idle"
	StateRunning      ServerState = "running"
	StateBusy         ServerState = "busy"
	StateStopping     ServerState = "stopping"
	StateStopped      ServerState = "stopped"
	StateStarting     ServerState = "starting"
	StateError        ServerState = "error"
	StateDisconnected ServerState = "disconnected"
	StateUnknown      ServerState = "unknown"
)

type ServerStats 

type ServerStats struct {
	RequestCount   int64
	ErrorCount     int64
	AvgLatencyMs   float64
	LastRequestAt  time.Time
	RestartCount   int
	CurrentBackoff time.Duration
	LastError      string
	LastErrorAt    time.Time
}

type StdioPool 

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

StdioPool manages multiple stdio-based MCP server subprocesses.

func NewStdioPool 

func NewStdioPool(maxPerServer int, idleTimeout time.Duration, logger *slog.Logger) *StdioPool

NewStdioPool creates a new StdioPool with the specified maximum servers per name and idle timeout.

func (*StdioPool) Close 

func (p *StdioPool) Close() error

func (*StdioPool) GetOrStartServer added in v0.6.1 

func (p *StdioPool) GetOrStartServer(ctx context.Context, name string) (*StdioServerV2, error)

func (*StdioPool) GetServer 

func (p *StdioPool) GetServer(name string) (*StdioServerV2, error)

func (*StdioPool) GetServerState 

func (p *StdioPool) GetServerState(name string) (ServerState, error)

func (*StdioPool) GetServerStats 

func (p *StdioPool) GetServerStats(name string) (ServerStats, error)

func (*StdioPool) HasServer added in v0.5.0 

func (p *StdioPool) HasServer(name string) bool

func (*StdioPool) IsServerMCPInitialized added in v0.7.1 

func (p *StdioPool) IsServerMCPInitialized(name string) bool

func (*StdioPool) ListServers 

func (p *StdioPool) ListServers() []string

func (*StdioPool) MarkServerMCPInitialized added in v0.7.1 

func (p *StdioPool) MarkServerMCPInitialized(name string)

func (*StdioPool) PutRequest 

func (p *StdioPool) PutRequest(name string, req Request) error

func (*StdioPool) RestartServer 

func (p *StdioPool) RestartServer(ctx context.Context, name string) error

func (*StdioPool) SendNotificationToServer added in v0.3.0 

func (p *StdioPool) SendNotificationToServer(ctx context.Context, name string, method string, params json.RawMessage) error

func (*StdioPool) SendRequest 

func (p *StdioPool) SendRequest(ctx context.Context, serverName string, req *proxy.JSONRPCRequest, timeout time.Duration) (*proxy.JSONRPCResponse, error)

func (*StdioPool) SendRequestToServer added in v0.3.0 

func (p *StdioPool) SendRequestToServer(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration) (*Response, error)

func (*StdioPool) SendRequestToServerWithID added in v0.3.0 

func (p *StdioPool) SendRequestToServerWithID(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration, id int) (*Response, error)

func (*StdioPool) SendServerNotification added in v0.3.0 

func (p *StdioPool) SendServerNotification(ctx context.Context, name string, method string, params map[string]interface{}) error

func (*StdioPool) ServerCount 

func (p *StdioPool) ServerCount() int

func (*StdioPool) SetReconnect added in v0.9.0 

func (p *StdioPool) SetReconnect(settings ReconnectSettings)

SetReconnect applies reconnect settings to the pool. It takes effect on servers already in the pool and on every server started afterwards.

func (*StdioPool) StartAllServers 

func (p *StdioPool) StartAllServers(ctx context.Context, configs []*migrate.ServerConfig) error

func (*StdioPool) StartServer 

func (p *StdioPool) StartServer(ctx context.Context, config *migrate.ServerConfig) error

func (*StdioPool) StopServer 

func (p *StdioPool) StopServer(name string) error

type StdioServerConfig 

type StdioServerConfig struct {
	Name            string
	Command         string
	Args            []string
	Env             []string
	CWD             string
	MaxConcurrent   int
	IdleTimeout     time.Duration
	RequestTimeout  time.Duration
	MaxResponseSize int
}

type StdioServerV2 

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

func (*StdioServerV2) IsMCPInitialized added in v0.7.1 

func (s *StdioServerV2) IsMCPInitialized() bool

func (*StdioServerV2) SetMCPInitialized added in v0.7.1 

func (s *StdioServerV2) SetMCPInitialized()

type UnifiedPool added in v0.5.0 

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

func NewUnifiedPool added in v0.5.0 

func NewUnifiedPool(stdio *StdioPool, http *HTTPClientPool, sse *SSEPool, logger *slog.Logger) *UnifiedPool

func (*UnifiedPool) Close added in v0.5.0 

func (p *UnifiedPool) Close() error

func (*UnifiedPool) GetServerState added in v0.5.0 

func (p *UnifiedPool) GetServerState(name string) (ServerState, error)

func (*UnifiedPool) IsServerMCPInitialized added in v0.7.1 

func (p *UnifiedPool) IsServerMCPInitialized(name string) bool

func (*UnifiedPool) ListServers added in v0.5.0 

func (p *UnifiedPool) ListServers() []string

func (*UnifiedPool) MarkServerMCPInitialized added in v0.7.1 

func (p *UnifiedPool) MarkServerMCPInitialized(name string)

func (*UnifiedPool) RestartServer added in v0.5.0 

func (p *UnifiedPool) RestartServer(ctx context.Context, name string) error

func (*UnifiedPool) SendRequest added in v0.5.0 

func (p *UnifiedPool) SendRequest(ctx context.Context, serverName string, req *proxy.JSONRPCRequest, timeout time.Duration) (*proxy.JSONRPCResponse, error)

func (*UnifiedPool) SendRequestToServer added in v0.5.0 

func (p *UnifiedPool) SendRequestToServer(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration) (*Response, error)

func (*UnifiedPool) SendRequestToServerWithID added in v0.5.0 

func (p *UnifiedPool) SendRequestToServerWithID(ctx context.Context, name string, method string, params json.RawMessage, timeout time.Duration, id int) (*Response, error)

func (*UnifiedPool) SendServerNotification added in v0.5.0 

func (p *UnifiedPool) SendServerNotification(ctx context.Context, name string, method string, params map[string]interface{}) error

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.