Add support for running the MCP server over HTTP with Server-Sent Events (SSE) using the MCP Streamable HTTP specification, alongside the existing STDIO transport. New features: - Transport abstraction with Transport interface - HTTP transport with session management - SSE support for server-initiated notifications - CORS security with configurable allowed origins - Optional TLS support - CLI flags for HTTP configuration (--transport, --http-address, etc.) - NixOS module options for HTTP transport The HTTP transport implements: - POST /mcp: JSON-RPC requests with session management - GET /mcp: SSE stream for server notifications - DELETE /mcp: Session termination - Origin validation (localhost-only by default) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
208 lines
4.1 KiB
Go
208 lines
4.1 KiB
Go
package mcp
|
|
|
|
import (
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// Session represents an MCP client session.
|
|
type Session struct {
|
|
ID string
|
|
CreatedAt time.Time
|
|
LastActivity time.Time
|
|
Initialized bool
|
|
|
|
// notifications is a channel for server-initiated notifications.
|
|
notifications chan *Response
|
|
|
|
mu sync.RWMutex
|
|
}
|
|
|
|
// NewSession creates a new session with a cryptographically secure random ID.
|
|
func NewSession() (*Session, error) {
|
|
id, err := generateSessionID()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
now := time.Now()
|
|
return &Session{
|
|
ID: id,
|
|
CreatedAt: now,
|
|
LastActivity: now,
|
|
notifications: make(chan *Response, 100),
|
|
}, nil
|
|
}
|
|
|
|
// Touch updates the session's last activity time.
|
|
func (s *Session) Touch() {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.LastActivity = time.Now()
|
|
}
|
|
|
|
// SetInitialized marks the session as initialized.
|
|
func (s *Session) SetInitialized() {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.Initialized = true
|
|
}
|
|
|
|
// IsInitialized returns whether the session has been initialized.
|
|
func (s *Session) IsInitialized() bool {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
return s.Initialized
|
|
}
|
|
|
|
// Notifications returns the channel for server-initiated notifications.
|
|
func (s *Session) Notifications() <-chan *Response {
|
|
return s.notifications
|
|
}
|
|
|
|
// SendNotification sends a notification to the session.
|
|
// Returns false if the channel is full.
|
|
func (s *Session) SendNotification(notification *Response) bool {
|
|
select {
|
|
case s.notifications <- notification:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// Close closes the session's notification channel.
|
|
func (s *Session) Close() {
|
|
close(s.notifications)
|
|
}
|
|
|
|
// SessionStore manages active sessions with TTL-based cleanup.
|
|
type SessionStore struct {
|
|
sessions map[string]*Session
|
|
ttl time.Duration
|
|
mu sync.RWMutex
|
|
stopClean chan struct{}
|
|
cleanDone chan struct{}
|
|
}
|
|
|
|
// NewSessionStore creates a new session store with the given TTL.
|
|
func NewSessionStore(ttl time.Duration) *SessionStore {
|
|
s := &SessionStore{
|
|
sessions: make(map[string]*Session),
|
|
ttl: ttl,
|
|
stopClean: make(chan struct{}),
|
|
cleanDone: make(chan struct{}),
|
|
}
|
|
go s.cleanupLoop()
|
|
return s
|
|
}
|
|
|
|
// Create creates a new session and adds it to the store.
|
|
func (s *SessionStore) Create() (*Session, error) {
|
|
session, err := NewSession()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.sessions[session.ID] = session
|
|
return session, nil
|
|
}
|
|
|
|
// Get retrieves a session by ID. Returns nil if not found or expired.
|
|
func (s *SessionStore) Get(id string) *Session {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
session, ok := s.sessions[id]
|
|
if !ok {
|
|
return nil
|
|
}
|
|
|
|
// Check if expired
|
|
session.mu.RLock()
|
|
expired := time.Since(session.LastActivity) > s.ttl
|
|
session.mu.RUnlock()
|
|
|
|
if expired {
|
|
return nil
|
|
}
|
|
|
|
return session
|
|
}
|
|
|
|
// Delete removes a session from the store.
|
|
func (s *SessionStore) Delete(id string) bool {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
session, ok := s.sessions[id]
|
|
if !ok {
|
|
return false
|
|
}
|
|
|
|
session.Close()
|
|
delete(s.sessions, id)
|
|
return true
|
|
}
|
|
|
|
// Count returns the number of active sessions.
|
|
func (s *SessionStore) Count() int {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
return len(s.sessions)
|
|
}
|
|
|
|
// Stop stops the cleanup goroutine and waits for it to finish.
|
|
func (s *SessionStore) Stop() {
|
|
close(s.stopClean)
|
|
<-s.cleanDone
|
|
}
|
|
|
|
// cleanupLoop periodically removes expired sessions.
|
|
func (s *SessionStore) cleanupLoop() {
|
|
defer close(s.cleanDone)
|
|
|
|
ticker := time.NewTicker(s.ttl / 2)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-s.stopClean:
|
|
return
|
|
case <-ticker.C:
|
|
s.cleanup()
|
|
}
|
|
}
|
|
}
|
|
|
|
// cleanup removes expired sessions.
|
|
func (s *SessionStore) cleanup() {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
now := time.Now()
|
|
for id, session := range s.sessions {
|
|
session.mu.RLock()
|
|
expired := now.Sub(session.LastActivity) > s.ttl
|
|
session.mu.RUnlock()
|
|
|
|
if expired {
|
|
session.Close()
|
|
delete(s.sessions, id)
|
|
}
|
|
}
|
|
}
|
|
|
|
// generateSessionID generates a cryptographically secure random session ID.
|
|
func generateSessionID() (string, error) {
|
|
bytes := make([]byte, 16)
|
|
if _, err := rand.Read(bytes); err != nil {
|
|
return "", err
|
|
}
|
|
return hex.EncodeToString(bytes), nil
|
|
}
|