server

package
v1.1.0 Latest Latest
Warning

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

Go to latest
Published: May 15, 2025 License: Apache-2.0 Imports: 14 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source
var (
	// ErrResponseWriterNotFlusher is returned when the ResponseWriter doesn't support Flusher interface
	ErrResponseWriterNotFlusher = errors.New("response writer does not implement http.Flusher")

	// ErrSessionNotFound is returned when a session cannot be found
	ErrSessionNotFound = errors.New("session not found")

	// ErrSessionClosed is returned when attempting to use a closed session
	ErrSessionClosed = errors.New("session is closed")

	// ErrChannelFull is returned when a notification channel is full
	ErrChannelFull = errors.New("notification channel is full")
)

Common errors in the server package

Functions

func NewSSEConnectionManager

func NewSSEConnectionManager() domain.ConnectionManager

NewSSEConnectionManager creates a new connection manager for SSE sessions.

func NewSSEHandler

func NewSSEHandler(
	config SSEHandlerConfig,
	messageHandler domain.MessageHandler,
	jsonrpcVersion string,
	notifier *NotificationSender,
) domain.SSEHandler

NewSSEHandler creates a new SSE handler with the given configuration and dependencies.

func NewSSESession

func NewSSESession(w http.ResponseWriter, userAgent string, bufferSize int) (domain.SSESession, error)

NewSSESession creates a new SSE session.

func NewTestServer

func NewTestServer(notifier *NotificationSender, mcpHandler func(ctx context.Context, rawMessage json.RawMessage) interface{}, opts ...SSEOption) *httptest.Server

NewTestServer creates a test server for testing purposes

Types

type ConnectionPool

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

ConnectionPool manages active SSE sessions.

func NewConnectionPool

func NewConnectionPool() *ConnectionPool

NewConnectionPool creates a new connection pool.

func (*ConnectionPool) Add

func (p *ConnectionPool) Add(session *sseSession)

Add adds a session to the pool.

func (*ConnectionPool) Broadcast

func (p *ConnectionPool) Broadcast(event interface{})

Broadcast sends an event to all active sessions.

func (*ConnectionPool) CloseAll

func (p *ConnectionPool) CloseAll()

CloseAll closes all active sessions.

func (*ConnectionPool) Count

func (p *ConnectionPool) Count() int

Count returns the number of active sessions.

func (*ConnectionPool) Get

func (p *ConnectionPool) Get(sessionID string) (*sseSession, bool)

Get returns a session by ID.

func (*ConnectionPool) Remove

func (p *ConnectionPool) Remove(sessionID string)

Remove removes a session from the pool.

type ContextKey

type ContextKey string

package-level constant for context key

const SessionIDContextKey ContextKey = "sessionId"

type InMemoryPromptRepository

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

InMemoryPromptRepository implements a PromptRepository using in-memory storage.

func NewInMemoryPromptRepository

func NewInMemoryPromptRepository() *InMemoryPromptRepository

NewInMemoryPromptRepository creates a new InMemoryPromptRepository.

func (*InMemoryPromptRepository) AddPrompt

func (r *InMemoryPromptRepository) AddPrompt(ctx context.Context, prompt *domain.Prompt) error

AddPrompt adds a new prompt to the repository.

func (*InMemoryPromptRepository) DeletePrompt

func (r *InMemoryPromptRepository) DeletePrompt(ctx context.Context, name string) error

DeletePrompt removes a prompt from the repository.

func (*InMemoryPromptRepository) GetPrompt

func (r *InMemoryPromptRepository) GetPrompt(ctx context.Context, name string) (*domain.Prompt, error)

GetPrompt retrieves a prompt by its name.

func (*InMemoryPromptRepository) ListPrompts

func (r *InMemoryPromptRepository) ListPrompts(ctx context.Context) ([]*domain.Prompt, error)

ListPrompts returns all available prompts.

type InMemoryResourceRepository

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

InMemoryResourceRepository implements a ResourceRepository using in-memory storage.

func NewInMemoryResourceRepository

func NewInMemoryResourceRepository() *InMemoryResourceRepository

NewInMemoryResourceRepository creates a new InMemoryResourceRepository.

func (*InMemoryResourceRepository) AddResource

func (r *InMemoryResourceRepository) AddResource(ctx context.Context, resource *domain.Resource) error

AddResource adds a new resource to the repository.

func (*InMemoryResourceRepository) DeleteResource

func (r *InMemoryResourceRepository) DeleteResource(ctx context.Context, uri string) error

DeleteResource removes a resource from the repository.

func (*InMemoryResourceRepository) GetResource

func (r *InMemoryResourceRepository) GetResource(ctx context.Context, uri string) (*domain.Resource, error)

GetResource retrieves a resource by its URI.

func (*InMemoryResourceRepository) ListResources

func (r *InMemoryResourceRepository) ListResources(ctx context.Context) ([]*domain.Resource, error)

ListResources returns all available resources.

type InMemorySessionRepository

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

InMemorySessionRepository implements a SessionRepository using in-memory storage.

func NewInMemorySessionRepository

func NewInMemorySessionRepository() *InMemorySessionRepository

NewInMemorySessionRepository creates a new InMemorySessionRepository.

func (*InMemorySessionRepository) AddSession

func (r *InMemorySessionRepository) AddSession(ctx context.Context, session *domain.ClientSession) error

AddSession adds a new session to the repository.

func (*InMemorySessionRepository) DeleteSession

func (r *InMemorySessionRepository) DeleteSession(ctx context.Context, id string) error

DeleteSession removes a session from the repository.

func (*InMemorySessionRepository) GetSession

GetSession retrieves a session by its ID.

func (*InMemorySessionRepository) ListSessions

ListSessions returns all active sessions.

type InMemoryToolRepository

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

InMemoryToolRepository implements a ToolRepository using in-memory storage.

func NewInMemoryToolRepository

func NewInMemoryToolRepository() *InMemoryToolRepository

NewInMemoryToolRepository creates a new InMemoryToolRepository.

func (*InMemoryToolRepository) AddTool

func (r *InMemoryToolRepository) AddTool(ctx context.Context, tool *domain.Tool) error

AddTool adds a new tool to the repository.

func (*InMemoryToolRepository) DeleteTool

func (r *InMemoryToolRepository) DeleteTool(ctx context.Context, name string) error

DeleteTool removes a tool from the repository.

func (*InMemoryToolRepository) GetTool

func (r *InMemoryToolRepository) GetTool(ctx context.Context, name string) (*domain.Tool, error)

GetTool retrieves a tool by its name.

func (*InMemoryToolRepository) ListTools

func (r *InMemoryToolRepository) ListTools(ctx context.Context) ([]*domain.Tool, error)

ListTools returns all available tools.

type JSONRPCNotification

type JSONRPCNotification struct {
	JSONRPC string                 `json:"jsonrpc"`
	Method  string                 `json:"method"`
	Params  map[string]interface{} `json:"params,omitempty"`
}

JSONRPCNotification represents a notification sent to clients via JSON-RPC.

type MCPSession

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

MCPSession represents a connected client session.

func NewMCPSession

func NewMCPSession(id, userAgent string, bufferSize int) *MCPSession

NewMCPSession creates a new MCPSession.

func (*MCPSession) Close

func (s *MCPSession) Close()

Close closes the notification channel.

func (*MCPSession) ID

func (s *MCPSession) ID() string

ID returns the session ID.

func (*MCPSession) NotificationChannel

func (s *MCPSession) NotificationChannel() NotificationChannel

NotificationChannel returns the channel for sending notifications to this session.

type NotificationChannel

type NotificationChannel chan JSONRPCNotification

NotificationChannel is a channel for sending notifications.

type NotificationRegistrar

type NotificationRegistrar interface {
	// RegisterSession registers a session for notifications.
	RegisterSession(session *MCPSession)

	// UnregisterSession unregisters a session.
	UnregisterSession(sessionID string)
}

NotificationRegistrar is an interface for registering and unregistering sessions.

type NotificationSender

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

NotificationSender handles sending notifications to clients.

func NewNotificationSender

func NewNotificationSender(jsonrpcVersion string) *NotificationSender

NewNotificationSender creates a new NotificationSender.

func (*NotificationSender) BroadcastNotification

func (n *NotificationSender) BroadcastNotification(ctx context.Context, notification *domain.Notification) error

BroadcastNotification sends a notification to all connected clients.

func (*NotificationSender) RegisterSession

func (n *NotificationSender) RegisterSession(session *MCPSession)

RegisterSession registers a session for notifications.

func (*NotificationSender) SendNotification

func (n *NotificationSender) SendNotification(ctx context.Context, sessionID string, notification *domain.Notification) error

SendNotification sends a notification to a specific client.

func (*NotificationSender) UnregisterSession

func (n *NotificationSender) UnregisterSession(sessionID string)

UnregisterSession unregisters a session.

type SSEContextFunc

type SSEContextFunc func(ctx context.Context, r *http.Request) context.Context

SSEContextFunc is a function that takes an existing context and the current request and returns a potentially modified context based on the request content. This can be used to inject context values from headers, for example.

type SSEHandlerConfig

type SSEHandlerConfig struct {
	BaseURL         string
	BasePath        string
	MessageEndpoint string
	SSEEndpoint     string
}

SSEHandlerConfig contains configuration options for the SSE handler.

type SSEOption

type SSEOption func(*SSEServer)

SSEOption defines a function type for configuring SSEServer

func WithBasePath

func WithBasePath(basePath string) SSEOption

WithBasePath sets the base path for the SSE server

func WithBaseURL

func WithBaseURL(baseURL string) SSEOption

WithBaseURL sets the base URL for the SSE server

func WithHTTPServer

func WithHTTPServer(srv *http.Server) SSEOption

WithHTTPServer sets the HTTP server instance

func WithLogger

func WithLogger(logger *logging.Logger) SSEOption

WithLogger sets the logger for the SSE server

func WithMessageEndpoint

func WithMessageEndpoint(endpoint string) SSEOption

WithMessageEndpoint sets the message endpoint path

func WithSSEContextFunc

func WithSSEContextFunc(fn SSEContextFunc) SSEOption

WithSSEContextFunc sets a function that will be called to customize the context to the server using the incoming request.

func WithSSEEndpoint

func WithSSEEndpoint(endpoint string) SSEOption

WithSSEEndpoint sets the SSE endpoint path

type SSEServer

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

SSEServer implements a Server-Sent Events (SSE) based server. It provides real-time communication capabilities over HTTP using the SSE protocol.

func NewSSEServer

func NewSSEServer(notifier *NotificationSender, mcpHandler func(ctx context.Context, rawMessage json.RawMessage) interface{}, opts ...SSEOption) *SSEServer

NewSSEServer creates a new SSE server instance with the given notification sender and options.

func (*SSEServer) BroadcastEvent

func (s *SSEServer) BroadcastEvent(event interface{})

BroadcastEvent sends an event to all active SSE sessions.

func (*SSEServer) CompleteMessageEndpoint

func (s *SSEServer) CompleteMessageEndpoint() string

func (*SSEServer) CompleteMessagePath

func (s *SSEServer) CompleteMessagePath() string

func (*SSEServer) CompleteSseEndpoint

func (s *SSEServer) CompleteSseEndpoint() string

func (*SSEServer) CompleteSsePath

func (s *SSEServer) CompleteSsePath() string

func (*SSEServer) GetUrlPath

func (s *SSEServer) GetUrlPath(input string) (string, error)

func (*SSEServer) SendEventToSession

func (s *SSEServer) SendEventToSession(
	sessionID string,
	event interface{},
) error

SendEventToSession sends an event to a specific SSE session identified by sessionID. Returns an error if the session is not found or closed.

func (*SSEServer) ServeHTTP

func (s *SSEServer) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP implements the http.Handler interface.

func (*SSEServer) Shutdown

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

Shutdown gracefully stops the SSE server, closing all active sessions and shutting down the HTTP server.

func (*SSEServer) Start

func (s *SSEServer) Start(addr string) error

Start begins serving SSE connections on the specified address. It sets up HTTP handlers for SSE and message endpoints.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL