aggregator

func (*AggregatorManager) GetAggregatorServer ¶

func (am *AggregatorManager) GetAggregatorServer() *AggregatorServer

GetAggregatorServer returns the underlying aggregator server instance.

This method provides access to advanced aggregator operations that are not exposed through the manager interface. It should be used carefully and primarily for testing or debugging purposes.

Returns nil if the server is not initialized.

func (*AggregatorManager) GetEndpoint ¶

func (am *AggregatorManager) GetEndpoint() string

GetEndpoint returns the aggregator's MCP endpoint URL.

The endpoint format depends on the configured transport:

• SSE: http://host:port/sse
• Streamable HTTP: http://host:port/mcp
• Stdio: "stdio"

Returns an empty string if the aggregator server is not available.

func (*AggregatorManager) GetEventHandler ¶

func (am *AggregatorManager) GetEventHandler() *EventHandler

GetEventHandler returns the event handler instance.

This method is primarily intended for testing and debugging purposes to inspect the state of the event handling system.

Returns nil if the event handler is not initialized.

func (*AggregatorManager) GetServiceData ¶

func (am *AggregatorManager) GetServiceData() map[string]interface{}

GetServiceData returns comprehensive service monitoring data.

This method provides detailed information about the aggregator's current state, including configuration, connection status, tool/resource/prompt counts, and server statistics. The data is suitable for monitoring dashboards and health checks.

Returns a map containing various metrics and status information.

func (*AggregatorManager) ManualRefresh ¶

func (am *AggregatorManager) ManualRefresh(ctx context.Context) error

ManualRefresh manually triggers a re-synchronization of all healthy MCP servers.

This method can be useful for debugging or when you need to force a refresh of the server registrations outside of the normal event-driven flow. It performs the same operation as the initial sync during startup.

Args:

• ctx: Context for the refresh operation

Returns an error if the refresh operation fails.

func (*AggregatorManager) RegisterServerPendingAuth ¶

func (am *AggregatorManager) RegisterServerPendingAuth(serverName, url, toolPrefix string, authInfo *AuthInfo) error

RegisterServerPendingAuth registers a server that requires authentication. This creates a placeholder with a synthetic auth tool that users can call to initiate the OAuth flow.

Args:

• serverName: Unique name of the server
• url: The server endpoint URL
• toolPrefix: Server-specific prefix for tools
• authInfo: OAuth information from the 401 response

Returns an error if registration fails.

func (*AggregatorManager) RegisterServerPendingAuthWithConfig ¶

func (am *AggregatorManager) RegisterServerPendingAuthWithConfig(serverName, url, toolPrefix string, authInfo *AuthInfo, authConfig *api.MCPServerAuth) error

RegisterServerPendingAuthWithConfig registers a server that requires authentication with auth config. This is an extended version that also accepts auth config for SSO token forwarding.

Args:

• serverName: Unique name of the server
• url: The server endpoint URL
• toolPrefix: Server-specific prefix for tools
• authInfo: OAuth information from the 401 response
• authConfig: Auth configuration for token forwarding (may be nil)

Returns an error if registration fails.

func (*AggregatorManager) Start ¶

func (am *AggregatorManager) Start(ctx context.Context) error

Start initializes and starts the aggregator manager.

This method performs the following initialization sequence:

1. Starts the underlying aggregator server
2. Validates that required APIs are available
3. Performs initial sync of healthy MCP servers
4. Sets up event handling for automatic updates
5. Starts periodic retry mechanism for failed registrations

The method is idempotent - calling it multiple times has no additional effect. Returns an error if any component fails to start.

func (*AggregatorManager) Stop ¶

func (am *AggregatorManager) Stop(ctx context.Context) error

Stop gracefully shuts down the aggregator manager.

This method stops all components in reverse order of startup:

1. Cancels the context to signal shutdown to all goroutines
2. Stops the event handler
3. Stops the OAuth manager
4. Stops the aggregator server
5. Waits for all background operations to complete

The method is idempotent and can be called multiple times safely.

func (*AggregatorManager) UpgradeServerAfterAuth ¶

func (am *AggregatorManager) UpgradeServerAfterAuth(ctx context.Context, serverName string, client MCPClient) error

UpgradeServerAfterAuth upgrades a pending auth server to connected status after successful OAuth authentication. This is called when the OAuth callback is received and a token is available.

Note: This method upgrades the global server state. For session-scoped tool visibility, use UpgradeSessionConnection instead.

Args:

• ctx: Context for the operation
• serverName: Name of the server to upgrade
• client: The newly authenticated MCP client

Returns an error if upgrade fails.

func (*AggregatorManager) UpgradeSessionConnection ¶

func (am *AggregatorManager) UpgradeSessionConnection(ctx context.Context, sessionID, serverName string, token oauth.RedactedToken) error

UpgradeSessionConnection upgrades a session's connection to an OAuth server after successful authentication. This creates a session-specific MCP client with the user's token and fetches their available tools.

This implements session-scoped tool visibility as described in ADR-006.

Args:

• ctx: Context for the operation
• sessionID: The session to upgrade
• serverName: The server the session authenticated with
• token: The OAuth access token (wrapped in RedactedToken for safety)

Returns an error if the upgrade fails.

func (*AggregatorServer) CallToolInternal ¶

func (a *AggregatorServer) CallToolInternal(ctx context.Context, toolName string, args map[string]interface{}) (*mcp.CallToolResult, error)

CallToolInternal provides internal tool calling capability for muster components.

This method allows internal muster components to execute tools through the aggregator without going through the external MCP protocol. It supports both:

• Tools from registered backend servers (resolved through the registry)
• Core tools from muster components (called directly through providers)

The method performs intelligent tool resolution:

1. First attempts to resolve the tool through the server registry
2. If not found, checks if it's a core tool by name pattern
3. Routes the call to the appropriate handler based on tool type

This internal calling mechanism is essential for:

• Inter-component communication within muster
• Workflow execution that needs to call other tools
• Administrative operations that require tool access

Args:

• ctx: Context for the tool execution
• toolName: Name of the tool to execute (may be prefixed)
• args: Arguments to pass to the tool as key-value pairs

Returns the tool execution result or an error if the tool cannot be found or executed.

func (*AggregatorServer) DeregisterServer ¶

func (a *AggregatorServer) DeregisterServer(name string) error

DeregisterServer removes a backend MCP server from the aggregator.

This method cleanly removes a backend server from the aggregator, which will cause all tools, resources, and prompts from that server to become unavailable. The backend client connection is closed as part of the deregistration process.

Additionally, this method cleans up any stale session connections for the server. This is critical for handling MCPServer renames, where the old server is deleted and a new one is created. Without this cleanup, session connections stored under the old server name would persist and cause stale auth status displays.

Args:

• name: Unique identifier of the server to remove

Returns an error if the server is not found or deregistration fails.

func (*AggregatorServer) GetAvailableTools ¶

func (a *AggregatorServer) GetAvailableTools() []string

GetAvailableTools implements the ToolAvailabilityChecker interface.

This method returns a comprehensive list of all tools currently available through the aggregator, including both external tools from backend servers and core tools from muster components. The returned list represents the complete tool inventory that can be used by workflows, capabilities, and other muster components.

The aggregation process:

1. Collects all tools from registered backend servers via the registry
2. Collects all core tools from muster component providers
3. Combines both lists into a unified tool inventory
4. Returns tool names (with appropriate prefixing applied)

This method is used by:

• Workflow manager for populating available tool lists
• Service class manager for tool validation
• Administrative interfaces for tool discovery

Returns a slice of tool names representing all available tools.

func (*AggregatorServer) GetEndpoint ¶

func (a *AggregatorServer) GetEndpoint() string

GetEndpoint returns the aggregator's primary endpoint URL based on the configured transport.

The endpoint format varies by transport type:

• SSE: http://host:port/sse (Server-Sent Events endpoint)
• Streamable HTTP: http://host:port/mcp (default HTTP streaming path)
• Stdio: "stdio" (indicates standard I/O communication)

This endpoint can be used by MCP clients to connect to the aggregator and access all aggregated capabilities from backend servers.

Returns the endpoint URL as a string, or "stdio" for standard I/O transport.

func (*AggregatorServer) GetPrompt ¶

func (a *AggregatorServer) GetPrompt(ctx context.Context, name string, args map[string]string) (*mcp.GetPromptResult, error)

GetPrompt executes a prompt with the provided arguments. This resolves the prompt name to its origin server and retrieves the prompt.

func (*AggregatorServer) GetPrompts ¶

func (a *AggregatorServer) GetPrompts() []mcp.Prompt

GetPrompts returns all available prompts from all sources with intelligent name prefixing.

This method aggregates prompts from all registered backend servers, applying smart prefixing similar to tools to avoid name conflicts. The prefixing ensures that prompts from different servers can coexist without naming collisions.

Returns a slice of MCP prompts ready for client consumption.

func (*AggregatorServer) GetPromptsForSession ¶

func (a *AggregatorServer) GetPromptsForSession(sessionID string) []mcp.Prompt

GetPromptsForSession returns a session-specific view of all available prompts.

Similar to GetToolsForSession, this returns prompts based on session authentication state.

func (*AggregatorServer) GetRegistry ¶

func (a *AggregatorServer) GetRegistry() *ServerRegistry

GetRegistry returns the server registry for direct access to backend server information.

This method provides access to the underlying registry for advanced operations such as inspecting server status, accessing raw capabilities, or performing administrative tasks. It should be used carefully to avoid disrupting the aggregator's normal operation.

Returns the ServerRegistry instance managing all backend servers.

func (*AggregatorServer) GetResources ¶

func (a *AggregatorServer) GetResources() []mcp.Resource

GetResources returns all available resources from all registered backend servers.

This method aggregates resources from all connected backend servers, applying appropriate URI prefixing to avoid conflicts. Resources maintain their original functionality while being properly namespaced within the aggregated environment.

Returns a slice of MCP resources ready for client access.

func (*AggregatorServer) GetResourcesForSession ¶

func (a *AggregatorServer) GetResourcesForSession(sessionID string) []mcp.Resource

GetResourcesForSession returns a session-specific view of all available resources.

Similar to GetToolsForSession, this returns resources based on session authentication state.

func (*AggregatorServer) GetSessionRegistry ¶

func (a *AggregatorServer) GetSessionRegistry() *SessionRegistry

GetSessionRegistry returns the session registry for per-session state management.

This method provides access to the session registry for managing per-session connections to OAuth-protected servers. It is used for session-scoped tool visibility and connection management.

Returns the SessionRegistry instance managing per-session state.

func (*AggregatorServer) GetTools ¶

func (a *AggregatorServer) GetTools() []mcp.Tool

GetTools returns all available tools from all sources with intelligent name prefixing.

This method aggregates tools from all registered backend servers, applying smart prefixing to avoid name conflicts. The prefixing is only applied when conflicts would otherwise occur, following the pattern: {muster_prefix}_{server_prefix}_{original_name}

Note: This returns the global tool view. For session-specific tool visibility, use GetToolsForSession instead.

Returns a slice of MCP tools ready for client consumption.

func (*AggregatorServer) GetToolsForSession ¶

func (a *AggregatorServer) GetToolsForSession(sessionID string) []mcp.Tool

GetToolsForSession returns a session-specific view of all available tools.

This method computes the tool view based on the session's authentication state:

• Global tools from servers that don't require authentication
• Tools from OAuth servers where the session has authenticated
• Synthetic auth tools for OAuth servers the session hasn't authenticated with

This implements per-session tool visibility as described in ADR-006.

Args:

• sessionID: The session to compute the tool view for

Returns a slice of MCP tools specific to this session.

func (*AggregatorServer) GetToolsWithStatus ¶

func (a *AggregatorServer) GetToolsWithStatus() []ToolWithStatus

GetToolsWithStatus returns all available tools along with their security blocking status.

This method provides enhanced tool information that includes whether each tool is blocked by the security denylist. The blocking status depends on:

• The tool's classification as destructive in the denylist
• The current yolo mode setting (yolo=true allows all tools)

The tool names are resolved to their original names (before prefixing) for accurate denylist checking, ensuring consistent security behavior regardless of how tools are exposed.

Returns a slice of ToolWithStatus containing both tool definitions and security status.

func (*AggregatorServer) IsToolAvailable ¶

func (a *AggregatorServer) IsToolAvailable(toolName string) bool

IsToolAvailable implements the ToolAvailabilityChecker interface.

This method determines whether a specific tool is available through the aggregator, checking both external backend servers (via the registry) and core muster tools (via name pattern matching). It provides a unified way for other components to verify tool availability before attempting to use them.

The check process:

1. Attempts to resolve the tool through the server registry
2. If not found, checks if it matches core tool naming patterns
3. Returns true if found in either location

This method is used by:

• Workflow manager for validating workflow step tools
• Service class manager for tool availability validation

Args:

• toolName: Name of the tool to check (may be prefixed)

Returns true if the tool is available, false otherwise.

func (*AggregatorServer) IsYoloMode ¶

func (a *AggregatorServer) IsYoloMode() bool

IsYoloMode returns whether yolo mode is currently enabled.

Yolo mode disables the security denylist, allowing all tools to be executed regardless of their destructive potential. This mode should only be enabled in development or testing environments where the risk is acceptable.

Returns true if yolo mode is enabled, false if security filtering is active.

func (*AggregatorServer) ListPromptsForContext ¶

func (a *AggregatorServer) ListPromptsForContext(ctx context.Context) []mcp.Prompt

ListPromptsForContext returns all available prompts for the current session context. This is used by the metatools package to provide session-scoped prompt visibility.

func (*AggregatorServer) ListResourcesForContext ¶

func (a *AggregatorServer) ListResourcesForContext(ctx context.Context) []mcp.Resource

ListResourcesForContext returns all available resources for the current session context. This is used by the metatools package to provide session-scoped resource visibility.

func (*AggregatorServer) ListServersRequiringAuth ¶

func (a *AggregatorServer) ListServersRequiringAuth(ctx context.Context) []api.ServerAuthInfo

ListServersRequiringAuth returns a list of servers that require authentication for the current session. This enables the list_tools meta-tool to inform users about servers that are available but require authentication before their tools become visible.

The method checks each registered server and returns those that:

• Have StatusAuthRequired status
• The session has not yet authenticated to

This is part of the server-side meta-tools migration (Issue #343) to provide better visibility into which servers need authentication.

func (*AggregatorServer) ListToolsForContext ¶

func (a *AggregatorServer) ListToolsForContext(ctx context.Context) []mcp.Tool

ListToolsForContext returns all available tools for the current session context. This is used by the metatools package to provide session-scoped tool visibility.

The method extracts the session ID from the context and returns tools appropriate for that session's authentication state. This includes:

• MCP server tools (prefixed with x_<server>_)
• Core muster tools (prefixed with core_) from internal providers

The core tools are collected from workflow, service, config, serviceclass, mcpserver, events, and auth providers.

func (*AggregatorServer) NotifySessionPromptsChanged ¶

func (a *AggregatorServer) NotifySessionPromptsChanged(sessionID string)

NotifySessionPromptsChanged sends a prompts/list_changed notification to a specific session.

func (*AggregatorServer) NotifySessionResourcesChanged ¶

func (a *AggregatorServer) NotifySessionResourcesChanged(sessionID string)

NotifySessionResourcesChanged sends a resources/list_changed notification to a specific session.

func (*AggregatorServer) NotifySessionToolsChanged ¶

func (a *AggregatorServer) NotifySessionToolsChanged(sessionID string)

NotifySessionToolsChanged sends a tools/list_changed notification to a specific session.

This method is used to notify a specific session that their tool list has changed, typically after they complete OAuth authentication with a new server. This implements targeted notifications as described in ADR-006, avoiding broadcast to all sessions.

Args:

• sessionID: The session to notify

func (*AggregatorServer) OnToolsUpdated ¶

func (a *AggregatorServer) OnToolsUpdated(event api.ToolUpdateEvent)

OnToolsUpdated implements the ToolUpdateSubscriber interface for handling tool change events.

This method responds to tool update events from other muster components, particularly the workflow manager, to maintain synchronization of the aggregator's exposed capabilities with the current tool landscape.

Event Processing:

• Filters events to focus on workflow-related tool changes
• Triggers capability refresh for workflow tool updates
• Uses asynchronous processing with a small delay to avoid mutex conflicts
• Logs all received events for debugging and monitoring

The asynchronous processing pattern ensures that:

• The event publisher (workflow manager) doesn't block waiting for aggregator updates
• Mutex conflicts are avoided by allowing the publisher to complete first
• Capability updates happen promptly but safely

Args:

• event: Tool update event containing change information, tool lists, and metadata

The method processes events selectively, focusing on workflow manager events that indicate changes to workflow-based tools.

func (*AggregatorServer) ReadResource ¶

func (a *AggregatorServer) ReadResource(ctx context.Context, uri string) (*mcp.ReadResourceResult, error)

ReadResource retrieves the contents of a resource by URI. This resolves the resource URI to its origin server and reads the content.

func (*AggregatorServer) RegisterServer ¶

func (a *AggregatorServer) RegisterServer(ctx context.Context, name string, client MCPClient, toolPrefix string) error

RegisterServer registers a new backend MCP server with the aggregator.

This method adds a backend MCP server to the aggregator's registry, making its tools, resources, and prompts available through the aggregated interface. The registration process includes client initialization and capability discovery.

Args:

• ctx: Context for the registration operation and capability queries
• name: Unique identifier for the server within the aggregator
• client: MCP client interface for communicating with the backend server
• toolPrefix: Server-specific prefix for name collision resolution

Returns an error if registration fails due to naming conflicts, client issues, or communication problems with the backend server.

func (*AggregatorServer) Start ¶

func (a *AggregatorServer) Start(ctx context.Context) error

Start initializes and starts the aggregator server with all configured transports.

This method performs a comprehensive startup sequence:

1. Creates and configures the core MCP server with full capabilities
2. Initializes workflow adapter if config directory is provided
3. Starts background monitoring of registry updates
4. Subscribes to tool update events from other muster components
5. Performs initial capability discovery and registration
6. Starts the appropriate transport server(s) based on configuration

Transport Support:

• SSE: Server-Sent Events with HTTP endpoints (/sse, /message)
• Stdio: Standard input/output for CLI integration
• Streamable HTTP: HTTP-based streaming protocol (default)

The method is idempotent for the same context - calling it multiple times with the same context will return an error indicating the server is already started.

Args:

• ctx: Context for controlling the server lifecycle and coordinating shutdown

Returns an error if startup fails at any stage, or nil on successful startup.

func (*AggregatorServer) Stop ¶

func (a *AggregatorServer) Stop(ctx context.Context) error

Stop gracefully shuts down the aggregator server and all its components.

This method performs a coordinated shutdown sequence:

1. Cancels the context to signal shutdown to all background routines
2. Shuts down all transport servers with a timeout
3. Waits for background routines to complete
4. Deregisters all backend servers to clean up connections
5. Resets internal state to allow for restart

The shutdown process includes:

• Graceful shutdown of HTTP-based transports with a 5-second timeout
• Automatic shutdown of stdio transport via context cancellation
• Cleanup of all registered backend MCP servers
• Synchronization with background monitoring routines

The method is idempotent - calling it multiple times is safe and will not cause errors or duplicate cleanup operations.

Args:

• ctx: Context for controlling the shutdown timeout and operations

Returns an error if shutdown encounters issues, though cleanup continues regardless.

func (*AggregatorServer) ToggleToolBlock ¶

func (a *AggregatorServer) ToggleToolBlock(toolName string) error

ToggleToolBlock toggles the blocked status of a specific tool (placeholder implementation).

This method is intended to provide runtime control over individual tool blocking, allowing administrators to override the default denylist behavior for specific tools. Currently, this functionality is not fully implemented and returns an error.

Future Enhancement: The full implementation would maintain a runtime override list that could selectively enable or disable specific tools regardless of the global yolo setting.

Args:

• toolName: Name of the tool to toggle blocking status for

Returns an error indicating the feature is not yet implemented.

func (*AggregatorServer) UpdateCapabilities ¶

func (a *AggregatorServer) UpdateCapabilities()

UpdateCapabilities provides public access to capability updates for external components.

This method exposes the internal updateCapabilities functionality to allow other muster components (particularly the workflow manager) to trigger capability refreshes when they detect changes in their tool inventory.

The method is thread-safe and can be called concurrently without causing issues. It performs the same comprehensive capability update as the internal method, including cleanup of obsolete items and addition of new capabilities.

Use Cases:

• Workflow manager triggering updates when workflow definitions change
• Administrative tools forcing capability refresh
• Integration testing scenarios requiring capability synchronization

This is a lightweight wrapper around the internal updateCapabilities method.

func (*AuthMetrics) GetServerMetrics ¶

func (m *AuthMetrics) GetServerMetrics(serverName string) (AuthServerMetricView, bool)

GetServerMetrics returns metrics for a specific server.

func (*AuthMetrics) GetSummary ¶

func (m *AuthMetrics) GetSummary() AuthMetricsSummary

GetSummary returns a summary of all authentication metrics.

func (*AuthMetrics) RecordLoginAttempt ¶

func (m *AuthMetrics) RecordLoginAttempt(serverName, sessionID string)

RecordLoginAttempt records an authentication login attempt.

Args:

• serverName: Name of the server being authenticated to
• sessionID: The session making the attempt (for logging, truncated)

func (*AuthMetrics) RecordLoginFailure ¶

func (m *AuthMetrics) RecordLoginFailure(serverName, sessionID, reason string)

RecordLoginFailure records a failed authentication attempt.

Args:

• serverName: Name of the server where authentication failed
• sessionID: The session that failed (for logging, truncated)
• reason: The reason for the failure

func (*AuthMetrics) RecordLoginSuccess ¶

func (m *AuthMetrics) RecordLoginSuccess(serverName, sessionID string)

RecordLoginSuccess records a successful authentication.

Args:

• serverName: Name of the server authenticated to
• sessionID: The session that authenticated (for logging, truncated)

func (*AuthMetrics) RecordLogoutAttempt ¶

func (m *AuthMetrics) RecordLogoutAttempt(serverName, sessionID string)

RecordLogoutAttempt records a logout attempt.

Args:

• serverName: Name of the server being logged out from
• sessionID: The session making the attempt (for logging, truncated)

func (*AuthMetrics) RecordLogoutSuccess ¶

func (m *AuthMetrics) RecordLogoutSuccess(serverName, sessionID string)

RecordLogoutSuccess records a successful logout.

Args:

• serverName: Name of the server logged out from
• sessionID: The session that logged out (for logging, truncated)

func (*AuthMetrics) RecordRateLimitBlock ¶

func (m *AuthMetrics) RecordRateLimitBlock(serverName, sessionID string)

RecordRateLimitBlock records when a session was rate limited.

Args:

• serverName: Name of the server the session was trying to authenticate to
• sessionID: The session that was blocked (for logging, truncated)

func (*AuthRateLimiter) Allow ¶

func (rl *AuthRateLimiter) Allow(sessionID, serverName string) bool

Allow checks if an auth attempt is allowed for the given session. If allowed, the attempt is recorded and true is returned. If rate limited, false is returned and the attempt is NOT recorded.

Args:

• sessionID: The session making the auth attempt
• serverName: The server being authenticated to (for logging)

Returns true if the attempt is allowed, false if rate limited.

func (*AuthRateLimiter) Cleanup ¶

func (rl *AuthRateLimiter) Cleanup()

Cleanup removes stale entries from the rate limiter. This is called periodically by the background cleanup goroutine.

func (*AuthRateLimiter) RemainingAttempts ¶

func (rl *AuthRateLimiter) RemainingAttempts(sessionID string) int

RemainingAttempts returns the number of remaining auth attempts for a session.

func (*AuthRateLimiter) Reset ¶

func (rl *AuthRateLimiter) Reset(sessionID string)

Reset clears all rate limiting state for a session. This can be called after successful authentication to reset the counter.

func (*AuthRateLimiter) Stop ¶ added in v0.1.34

func (rl *AuthRateLimiter) Stop()

Stop terminates the background cleanup goroutine. It is safe to call Stop multiple times.

func (*AuthToolProvider) ExecuteTool ¶

func (p *AuthToolProvider) ExecuteTool(ctx context.Context, toolName string, args map[string]interface{}) (*api.CallToolResult, error)

ExecuteTool executes an authentication tool by name.

func (*ConnectionAlreadyEstablishedError) Error ¶

func (e *ConnectionAlreadyEstablishedError) Error() string

func (*ConnectionNotFoundError) Error ¶

func (e *ConnectionNotFoundError) Error() string

func (*EventHandler) IsRunning ¶

func (eh *EventHandler) IsRunning() bool

IsRunning returns whether the event handler is currently active.

This method is thread-safe and can be used to check the handler's status for monitoring or debugging purposes.

Returns true if the event handler is currently processing events.

func (*EventHandler) Start ¶

func (eh *EventHandler) Start(ctx context.Context) error

Start begins listening for orchestrator events and processing them asynchronously.

This method subscribes to service state change events from the orchestrator and starts a background goroutine to process them. The method is idempotent - calling it multiple times has no additional effect.

The event processing continues until the provided context is cancelled or the Stop method is called.

Args:

• ctx: Context for controlling the event handler lifecycle

Returns nil on successful startup. The method does not wait for event processing to complete, as that happens asynchronously.

func (*EventHandler) Stop ¶

func (eh *EventHandler) Stop() error

Stop gracefully shuts down the event handler and waits for cleanup to complete.

This method cancels the event processing context and waits for the background goroutine to finish processing any in-flight events. The method is idempotent and can be called multiple times safely.

Returns nil after successful shutdown.

func (*InvalidSessionIDError) Error ¶

func (e *InvalidSessionIDError) Error() string

func (*NameTracker) GetExposedPromptName ¶

func (nt *NameTracker) GetExposedPromptName(serverName, promptName string) string

GetExposedPromptName returns the fully prefixed name for a prompt.

This method applies both server-specific and muster prefixes to create a globally unique prompt name. It also maintains the internal mapping for later resolution.

The naming pattern is: {muster_prefix}_{server_prefix}_{original_name}

Args:

• serverName: Name of the server providing the prompt
• promptName: Original prompt name from the server

Returns the exposed prompt name that clients will see.

func (*NameTracker) GetExposedResourceURI ¶

func (nt *NameTracker) GetExposedResourceURI(serverName, resourceURI string) string

GetExposedResourceURI returns the fully prefixed URI for a resource.

This method handles resource URI prefixing with special consideration for URIs that already contain schemes (like http:// or file://). Such URIs are left unchanged to preserve their functionality.

For simple resource identifiers, the standard prefixing pattern is applied.

Args:

• serverName: Name of the server providing the resource
• resourceURI: Original resource URI from the server

Returns the exposed resource URI that clients will see.

func (*NameTracker) GetExposedToolName ¶

func (nt *NameTracker) GetExposedToolName(serverName, toolName string) string

GetExposedToolName returns the fully prefixed name for a tool.

This method applies both server-specific and muster prefixes to create a globally unique tool name. It also maintains the internal mapping for later resolution.

The naming pattern is: {muster_prefix}_{server_prefix}_{original_name}

Args:

• serverName: Name of the server providing the tool
• toolName: Original tool name from the server

Returns the exposed tool name that clients will see.

func (*NameTracker) RebuildMappings ¶

func (nt *NameTracker) RebuildMappings(servers map[string]*ServerInfo)

RebuildMappings is a legacy method maintained for backward compatibility.

In the current implementation, name mappings are maintained automatically as capabilities are registered, so this method is no longer needed. It is kept to avoid breaking existing code that might call it.

Args:

• servers: Map of server information (unused in current implementation)

func (*NameTracker) ResolveName ¶

func (nt *NameTracker) ResolveName(exposedName string) (serverName, originalName string, itemType string, err error)

ResolveName resolves an exposed (prefixed) name back to its server and original name.

This method performs the reverse operation of the Get*Name methods, taking an exposed name and determining which server provided it and what the original name was before prefixing.

This is essential for routing requests to the correct backend server.

Args:

• exposedName: The prefixed name as seen by clients

Returns the server name, original name, item type, and nil error if successful. Returns empty strings and an error if the name cannot be resolved.

func (*NameTracker) SetServerPrefix ¶

func (nt *NameTracker) SetServerPrefix(serverName, prefix string)

SetServerPrefix configures the prefix to use for a specific server.

Server prefixes allow customization of how capabilities from different servers are distinguished. If no prefix is configured for a server, the server name itself is used as the prefix.

This method is thread-safe and can be called while the tracker is in use.

Args:

• serverName: Unique identifier of the server
• prefix: Prefix to use for this server's capabilities (uses server name if empty)

func (*ServerInfo) GetNamespace ¶

func (s *ServerInfo) GetNamespace() string

GetNamespace returns the namespace for this server. Returns "default" if the namespace is not set.

func (*ServerInfo) IsConnected ¶

func (s *ServerInfo) IsConnected() bool

IsConnected returns the current connection status. This method is thread-safe and can be called concurrently.

func (*ServerInfo) SetConnected ¶

func (s *ServerInfo) SetConnected(connected bool)

SetConnected safely updates the connection status. This is used to track whether the server is currently available for operations.

func (*ServerInfo) UpdatePrompts ¶

func (s *ServerInfo) UpdatePrompts(prompts []mcp.Prompt)

UpdatePrompts safely updates the server's cached prompt list. This method is thread-safe and should be used whenever the server's available prompts change.

func (*ServerInfo) UpdateResources ¶

func (s *ServerInfo) UpdateResources(resources []mcp.Resource)

UpdateResources safely updates the server's cached resource list. This method is thread-safe and should be used whenever the server's available resources change.

func (*ServerInfo) UpdateTools ¶

func (s *ServerInfo) UpdateTools(tools []mcp.Tool)

UpdateTools safely updates the server's cached tool list. This method is thread-safe and should be used whenever the server's available tools change.

func (*ServerRegistry) Deregister ¶

func (r *ServerRegistry) Deregister(name string) error

Deregister removes an MCP server from the registry and cleans up its resources.

This method safely removes a server from the registry, closes its client connection, and notifies subscribers of the change. All tools, resources, and prompts provided by the server will no longer be available through the aggregator.

The method is thread-safe and can be called concurrently.

Args:

• name: Unique identifier of the server to remove

Returns an error if the server is not found in the registry.

func (*ServerRegistry) GetAllPrompts ¶

func (r *ServerRegistry) GetAllPrompts() []mcp.Prompt

GetAllPrompts returns a consolidated list of all prompts from all connected servers.

This method aggregates prompts from all registered and connected servers, applying intelligent prefixing to avoid name conflicts. Only servers that are currently connected contribute their prompts to the result.

Returns a slice of MCP prompts ready for exposure through the aggregator.

func (*ServerRegistry) GetAllPromptsForSession ¶

func (r *ServerRegistry) GetAllPromptsForSession(sessionRegistry *SessionRegistry, sessionID string) []mcp.Prompt

GetAllPromptsForSession returns a session-specific view of all available prompts.

Similar to GetAllToolsForSession, this returns prompts based on session authentication state.

func (*ServerRegistry) GetAllResources ¶

func (r *ServerRegistry) GetAllResources() []mcp.Resource

GetAllResources returns a consolidated list of all resources from all connected servers.

This method aggregates resources from all registered and connected servers, applying intelligent prefixing to resource URIs to avoid conflicts. Only servers that are currently connected contribute their resources to the result.

Returns a slice of MCP resources ready for exposure through the aggregator.

func (*ServerRegistry) GetAllResourcesForSession ¶

func (r *ServerRegistry) GetAllResourcesForSession(sessionRegistry *SessionRegistry, sessionID string) []mcp.Resource

GetAllResourcesForSession returns a session-specific view of all available resources.

Similar to GetAllToolsForSession, this returns resources based on session authentication state.

func (*ServerRegistry) GetAllServers ¶

func (r *ServerRegistry) GetAllServers() map[string]*ServerInfo

GetAllServers returns a copy of all registered server information.

This method provides a snapshot of all servers currently registered with the registry, including both connected and disconnected servers. The returned map is a copy to prevent external modifications to the internal state.

Returns a map of server names to their corresponding ServerInfo structures.

func (*ServerRegistry) GetAllTools ¶

func (r *ServerRegistry) GetAllTools() []mcp.Tool

GetAllTools returns a consolidated list of all tools from all connected servers.

This method aggregates tools from all registered and connected servers, applying intelligent prefixing to avoid name conflicts. Only servers that are currently connected contribute their tools to the result.

Per ADR-008, servers in auth_required state do NOT contribute any tools. Users must use core_auth_login to authenticate before server tools become visible.

The returned tools have their names modified to include appropriate prefixes following the pattern: {muster_prefix}_{server_prefix}_{original_name}

Returns a slice of MCP tools ready for exposure through the aggregator.

func (*ServerRegistry) GetAllToolsForSession ¶

func (r *ServerRegistry) GetAllToolsForSession(sessionRegistry *SessionRegistry, sessionID string) []mcp.Tool

GetAllToolsForSession returns a session-specific view of all available tools.

The returned tool list is computed based on the session's authentication state:

• GlobalTools: Tools from servers that don't require authentication
• AuthenticatedServerTools: Tools from OAuth servers where the session has a valid connection

Per ADR-008, servers requiring authentication but not yet authenticated do NOT show any tools. Users must use core_auth_login to authenticate before server tools become visible.

This implements per-session tool visibility as described in ADR-006 and ADR-008.

Args:

• sessionRegistry: The session registry containing per-session state
• sessionID: The session to compute the tool view for

Returns a slice of MCP tools specific to this session.

func (*ServerRegistry) GetClient ¶

func (r *ServerRegistry) GetClient(name string) (MCPClient, error)

GetClient returns the MCP client for a specific registered server.

This method provides access to the underlying MCP client for direct communication with a specific server. The client can be used to execute tools, read resources, or retrieve prompts from the server.

Args:

• name: Unique identifier of the server

Returns the MCP client interface and nil error if successful. Returns nil client and an error if the server is not found or not connected.

func (*ServerRegistry) GetOAuthServers ¶

func (r *ServerRegistry) GetOAuthServers() []*ServerInfo

GetOAuthServers returns a list of servers that require OAuth authentication.

func (*ServerRegistry) GetServerInfo ¶

func (r *ServerRegistry) GetServerInfo(name string) (*ServerInfo, bool)

GetServerInfo returns detailed information about a specific registered server.

This method provides access to the complete ServerInfo structure for a given server, including its client, cached capabilities, and connection status.

Args:

• name: Unique identifier of the server

Returns the ServerInfo pointer and true if the server exists. Returns nil and false if the server is not found.

func (*ServerRegistry) GetUpdateChannel ¶

func (r *ServerRegistry) GetUpdateChannel() <-chan struct{}

GetUpdateChannel returns a read-only channel that receives notifications when the registry is updated.

Subscribers can use this channel to react to server registrations, deregistrations, or capability changes. The channel is buffered with a capacity of 1 to prevent blocking the registry operations.

Returns a receive-only channel for registry update notifications.

func (*ServerRegistry) IsOAuthServer ¶

func (r *ServerRegistry) IsOAuthServer(serverName string) bool

IsOAuthServer checks if a server requires OAuth authentication.

func (*ServerRegistry) Register ¶

func (r *ServerRegistry) Register(ctx context.Context, name string, client MCPClient, toolPrefix string) error

Register adds a new MCP server to the registry and initializes its capabilities.

This method performs the following operations:

1. Validates that the server name is not already in use
2. Initializes the MCP client if needed
3. Queries the server for its initial capabilities
4. Stores the server information and updates the name tracker
5. Notifies subscribers of the registry update

The method is thread-safe and can be called concurrently.

Args:

• ctx: Context for initialization and capability queries
• name: Unique identifier for the server
• client: MCP client instance for communicating with the server
• toolPrefix: Server-specific prefix for tools (uses server name if empty)

Returns an error if the server name is already registered, client initialization fails, or the server cannot be reached.

func (*ServerRegistry) RegisterPendingAuth ¶

func (r *ServerRegistry) RegisterPendingAuth(name, url, toolPrefix string, authInfo *AuthInfo) error

RegisterPendingAuth registers a server that requires authentication before it can be fully connected. This creates a placeholder server entry with StatusAuthRequired. Per ADR-008, no synthetic authentication tools are created - users should use core_auth_login to authenticate.

Args:

• name: Unique identifier for the server
• url: The server endpoint URL
• toolPrefix: Server-specific prefix for tools
• authInfo: OAuth authentication information from the 401 response

Returns an error if the server name is already registered.

func (*ServerRegistry) RegisterPendingAuthWithConfig ¶

func (r *ServerRegistry) RegisterPendingAuthWithConfig(name, url, toolPrefix string, authInfo *AuthInfo, authConfig *api.MCPServerAuth) error

RegisterPendingAuthWithConfig registers a server that requires authentication with auth configuration. This is an extended version of RegisterPendingAuth that also accepts auth config for SSO token forwarding.

Args:

• name: Unique identifier for the server
• url: The server endpoint URL
• toolPrefix: Server-specific prefix for tools
• authInfo: OAuth authentication information from the 401 response
• authConfig: Auth configuration for token forwarding (may be nil)

Returns an error if the server name is already registered.

func (*ServerRegistry) ResolvePromptName ¶

func (r *ServerRegistry) ResolvePromptName(exposedName string) (serverName, originalName string, err error)

ResolvePromptName resolves an exposed (prefixed) prompt name back to its source server and original name.

This method is used when a prompt request is received to determine which server should handle the request and what the original prompt name was before prefixing.

Args:

• exposedName: The prefixed prompt name as seen by clients

Returns the server name, original prompt name, and nil error if resolution succeeds. Returns empty strings and an error if the name cannot be resolved or refers to a different item type.

func (*ServerRegistry) ResolveResourceName ¶

func (r *ServerRegistry) ResolveResourceName(exposedURI string) (serverName, originalURI string, err error)

ResolveResourceName resolves an exposed (prefixed) resource URI back to its source server and original URI.

This method is used when a resource read request is received to determine which server should handle the request and what the original resource URI was before prefixing.

Args:

• exposedURI: The prefixed resource URI as seen by clients

Returns the server name, original resource URI, and nil error if resolution succeeds. Returns empty strings and an error if the URI cannot be resolved or refers to a different item type.

func (*ServerRegistry) ResolveToolName ¶

func (r *ServerRegistry) ResolveToolName(exposedName string) (serverName, originalName string, err error)

ResolveToolName resolves an exposed (prefixed) tool name back to its source server and original name.

This method is used when a tool call is received to determine which server should handle the request and what the original tool name was before prefixing.

Args:

• exposedName: The prefixed tool name as seen by clients

Returns the server name, original tool name, and nil error if resolution succeeds. Returns empty strings and an error if the name cannot be resolved or refers to a different item type.

func (*ServerRegistry) UpgradeToConnected ¶

func (r *ServerRegistry) UpgradeToConnected(ctx context.Context, name string, client MCPClient) error

UpgradeToConnected upgrades a pending auth server to connected status after successful authentication. This replaces the synthetic auth tool with the server's actual tools.

Args:

• ctx: Context for capability queries
• name: Server name to upgrade
• client: The newly authenticated MCP client

Returns an error if the server is not found or is not in pending auth state.

func (*SessionConnection) GetAuthStatus ¶

func (sc *SessionConnection) GetAuthStatus() AuthStatus

GetAuthStatus returns the authentication status for a session connection.

func (*SessionConnection) GetConnectedAt ¶

func (sc *SessionConnection) GetConnectedAt() time.Time

GetConnectedAt returns the timestamp when the connection was established. This provides thread-safe access to the ConnectedAt field.

func (*SessionConnection) GetLastError ¶

func (sc *SessionConnection) GetLastError() string

GetLastError returns the last error for this session connection.

func (*SessionConnection) GetPrompts ¶

func (sc *SessionConnection) GetPrompts() []mcp.Prompt

GetPrompts returns the cached prompts for a session connection.

func (*SessionConnection) GetResources ¶

func (sc *SessionConnection) GetResources() []mcp.Resource

GetResources returns the cached resources for a session connection.

func (*SessionConnection) GetStatus ¶

func (sc *SessionConnection) GetStatus() ConnectionStatus

GetStatus returns the connection status for a session connection. This provides thread-safe access to the Status field.

func (*SessionConnection) GetTools ¶

func (sc *SessionConnection) GetTools() []mcp.Tool

GetTools returns the cached tools for a session connection.

func (*SessionConnection) IsTokenExpired ¶ added in v0.1.33

func (sc *SessionConnection) IsTokenExpired(margin time.Duration) bool

IsTokenExpired returns true if the connection's SSO token is expired or will expire within the given margin. Returns false if no expiry time is set (e.g., non-SSO connections or unknown expiry).

func (*SessionConnection) SetAuthStatus ¶

func (sc *SessionConnection) SetAuthStatus(status AuthStatus)

SetAuthStatus updates the authentication status for a session connection.

func (*SessionConnection) SetAuthStatusWithError ¶

func (sc *SessionConnection) SetAuthStatusWithError(status AuthStatus, errMsg string)

SetAuthStatusWithError updates the auth status and records an error message. This is used when authentication fails (e.g., token expired, invalid credentials).

func (*SessionConnection) SetTokenExpiresAt ¶ added in v0.1.35

func (sc *SessionConnection) SetTokenExpiresAt(t time.Time)

SetTokenExpiresAt updates the token expiry time. This is called when a dynamic header function successfully re-exchanges a token, so that the re-init guard sees the updated expiry instead of the stale initial value.

func (*SessionConnection) UpdatePrompts ¶

func (sc *SessionConnection) UpdatePrompts(prompts []mcp.Prompt)

UpdatePrompts updates the cached prompts for a session connection.

func (*SessionConnection) UpdateResources ¶

func (sc *SessionConnection) UpdateResources(resources []mcp.Resource)

UpdateResources updates the cached resources for a session connection.

func (*SessionConnection) UpdateTools ¶

func (sc *SessionConnection) UpdateTools(tools []mcp.Tool)

UpdateTools updates the cached tools for a session connection.

func (*SessionConnectionResult) FormatAsAPIResult ¶

func (r *SessionConnectionResult) FormatAsAPIResult() *api.CallToolResult

FormatAsAPIResult formats the connection result as an api.CallToolResult. Used by AuthToolProvider.tryConnectWithToken.

func (*SessionConnectionResult) FormatAsMCPResult ¶

func (r *SessionConnectionResult) FormatAsMCPResult() *mcp.CallToolResult

FormatAsMCPResult formats the connection result as an mcp.CallToolResult. Used by AggregatorServer.tryConnectWithToken.

func (*SessionIdentityMismatchError) Error ¶ added in v0.1.27

func (e *SessionIdentityMismatchError) Error() string

func (*SessionLimitExceededError) Error ¶

func (e *SessionLimitExceededError) Error() string

func (*SessionNotFoundError) Error ¶

func (e *SessionNotFoundError) Error() string

func (*SessionRegistry) Count ¶

func (sr *SessionRegistry) Count() int

Count returns the number of active sessions.

func (*SessionRegistry) CreateSessionForSubject ¶ added in v0.1.27

func (sr *SessionRegistry) CreateSessionForSubject(subject string) (*SessionState, error)

CreateSessionForSubject generates a new server-side session ID and binds it to the given subject (authenticated user identity). This is the primary entry point for server-side session creation.

Returns the new session state and session ID, or an error if generation fails or the session limit is exceeded.

func (*SessionRegistry) DeleteSession ¶

func (sr *SessionRegistry) DeleteSession(sessionID string)

DeleteSession removes a session and cleans up its connections.

This method closes all session-specific MCP client connections and removes the session state from the registry.

Args:

• sessionID: Unique identifier for the session to remove

func (*SessionRegistry) EndSSOInit ¶

func (sr *SessionRegistry) EndSSOInit(sessionID string)

EndSSOInit marks that SSO initialization has completed for a session. The session can now be cleaned up by timeout if it becomes idle.

func (*SessionRegistry) GetAllSessions ¶

func (sr *SessionRegistry) GetAllSessions() map[string]*SessionState

GetAllSessions returns all active sessions.

Returns a map of session IDs to session states. The returned map is a copy to prevent external modifications to the internal state.

func (*SessionRegistry) GetAuthStatus ¶

func (sr *SessionRegistry) GetAuthStatus(sessionID, serverName string) (AuthStatus, bool)

GetAuthStatus returns the authentication status for a session's connection to a server. This is per-user session state used to track if a user is authenticated, needs auth, etc.

Args:

• sessionID: Unique identifier for the session
• serverName: Name of the server

Returns the auth status and true if found, AuthStatusUnknown and false otherwise.

func (*SessionRegistry) GetConnection ¶

func (sr *SessionRegistry) GetConnection(sessionID, serverName string) (*SessionConnection, bool)

GetConnection returns a session's connection to a specific server.

Args:

• sessionID: Unique identifier for the session
• serverName: Name of the server

Returns the connection and true if found, nil and false otherwise.

func (*SessionRegistry) GetOrCreateSession ¶

func (sr *SessionRegistry) GetOrCreateSession(sessionID string) *SessionState

GetOrCreateSession returns the session state for a session ID, creating it if necessary.

This method is the primary entry point for session management. It ensures that a session always exists when needed and updates the last activity timestamp.

The method validates the session ID and enforces session limits:

• Empty or excessively long session IDs are rejected
• New sessions are rejected if the maximum session limit is reached

Args:

• sessionID: Unique identifier for the session

Returns the session state, or nil if validation fails or limits are exceeded. Callers should use GetOrCreateSessionWithError for detailed error information.

func (*SessionRegistry) GetOrCreateSessionWithError ¶

func (sr *SessionRegistry) GetOrCreateSessionWithError(sessionID string) (*SessionState, error)

GetOrCreateSessionWithError returns the session state for a session ID, creating it if necessary.

This method provides detailed error information for validation and limit failures.

Args:

• sessionID: Unique identifier for the session

Returns:

• The session state (nil on error)
• An error if validation fails or limits are exceeded

func (*SessionRegistry) GetSession ¶

func (sr *SessionRegistry) GetSession(sessionID string) (*SessionState, bool)

GetSession returns the session state for a session ID.

Args:

• sessionID: Unique identifier for the session

Returns the session state and true if found, nil and false otherwise. Invalid session IDs return nil and false.

func (*SessionRegistry) HasSSOFailed ¶

func (sr *SessionRegistry) HasSSOFailed(sessionID, serverName string) bool

HasSSOFailed checks if SSO authentication was attempted but failed for a server. Returns true if SSO was attempted and failed, false otherwise.

func (*SessionRegistry) IsSSOInitInProgress ¶

func (sr *SessionRegistry) IsSSOInitInProgress(sessionID string) bool

IsSSOInitInProgress checks if SSO initialization is in progress for a session.

func (*SessionRegistry) MarkSSOFailed ¶

func (sr *SessionRegistry) MarkSSOFailed(sessionID, serverName string)

MarkSSOFailed records that SSO authentication was attempted but failed for a server. This is called when proactive SSO (token forwarding) fails during session init. The failure is tracked so the UI can show "SSO failed" to explain why auth is required.

func (*SessionRegistry) RemoveServerFromAllSessions ¶

func (sr *SessionRegistry) RemoveServerFromAllSessions(serverName string) int

RemoveServerFromAllSessions removes connections to a specific server from all sessions. This is called when an MCPServer is deregistered (e.g., deleted or renamed) to clean up stale session connections that would otherwise persist until session timeout.

This fixes the issue where renamed MCPServers would show stale auth status because session connections were stored under the old server name and never cleaned up.

Args:

• serverName: The name of the server to remove from all sessions

Returns the number of connections that were removed.

func (*SessionRegistry) SetAuthStatus ¶

func (sr *SessionRegistry) SetAuthStatus(sessionID, serverName string, status AuthStatus)

SetAuthStatus updates the authentication status for a session's connection to a server.

Args:

• sessionID: Unique identifier for the session
• serverName: Name of the server
• status: New auth status

func (*SessionRegistry) SetAuthStatusWithError ¶

func (sr *SessionRegistry) SetAuthStatusWithError(sessionID, serverName string, status AuthStatus, errMsg string)

SetAuthStatusWithError updates the auth status and records an error for a session. This is used when authentication fails (e.g., token expired, invalid credentials).

Args:

• sessionID: Unique identifier for the session
• serverName: Name of the server
• status: New auth status
• errMsg: Error message describing the failure

func (*SessionRegistry) SetConnection ¶

func (sr *SessionRegistry) SetConnection(sessionID, serverName string, conn *SessionConnection)

SetConnection sets a session's connection to a specific server.

Args:

• sessionID: Unique identifier for the session
• serverName: Name of the server
• conn: The connection to set

func (*SessionRegistry) SetOnSessionRemoved ¶ added in v0.1.34

func (sr *SessionRegistry) SetOnSessionRemoved(fn func(sessionID string))

SetOnSessionRemoved registers a callback that is invoked whenever a session is removed from the registry (via cleanup or DeleteSession). This must be called before sessions are created (typically right after construction).

func (*SessionRegistry) SetPendingAuth ¶

func (sr *SessionRegistry) SetPendingAuth(sessionID, serverName string)

SetPendingAuth marks a server as pending authentication for a session.

This creates a placeholder connection in pending_auth state that will be upgraded to connected once authentication succeeds.

Args:

• sessionID: Unique identifier for the session
• serverName: Name of the server

func (*SessionRegistry) StartSSOInit ¶

func (sr *SessionRegistry) StartSSOInit(sessionID string)

StartSSOInit marks that SSO initialization is starting for a session. While SSO init is in progress, the session will not be cleaned up by timeout. This prevents race conditions where sessions are cleaned up before SSO completes.

IMPORTANT: Callers MUST call EndSSOInit when SSO initialization completes, typically using defer. Failure to do so will prevent the session from ever being cleaned up.

func (*SessionRegistry) Stop ¶

func (sr *SessionRegistry) Stop()

Stop stops the session registry and cleans up all sessions.

This method closes all session-specific MCP client connections, invokes the onSessionRemoved callback for each session, and stops the background cleanup goroutine.

func (*SessionRegistry) UpgradeConnection ¶

func (sr *SessionRegistry) UpgradeConnection(sessionID, serverName string, client MCPClient, tokenKey *oauth.TokenKey) error

UpgradeConnection upgrades a session's connection from pending_auth to connected.

This is called after successful OAuth authentication to associate the authenticated MCP client with the session.

Args:

• sessionID: Unique identifier for the session
• serverName: Name of the server
• client: The authenticated MCP client
• tokenKey: Reference to the token in the token store

Returns an error if the session or connection is not found.

func (*SessionRegistry) ValidateSessionIdentity ¶ added in v0.1.27

func (sr *SessionRegistry) ValidateSessionIdentity(sessionID, subject string) error

ValidateSessionIdentity checks that the session's bound identity matches the given subject. Returns nil if the session has no bound identity (legacy/unbound) or if it matches. Returns a SessionIdentityMismatchError if the identities don't match.

func (*SessionState) CloseAllConnections ¶

func (s *SessionState) CloseAllConnections()

CloseAllConnections closes all MCP client connections for this session.

func (*SessionState) GetAllConnections ¶

func (s *SessionState) GetAllConnections() map[string]*SessionConnection

GetAllConnections returns a copy of all connections for this session. This is safe for iteration since it returns a copy of the map.

func (*SessionState) GetConnectedServers ¶

func (s *SessionState) GetConnectedServers() []string

GetConnectedServers returns the names of all servers the session is connected to.

func (*SessionState) GetConnection ¶

func (s *SessionState) GetConnection(serverName string) (*SessionConnection, bool)

GetConnection returns the session's connection to a specific server.

func (*SessionState) GetPendingAuthServers ¶

func (s *SessionState) GetPendingAuthServers() []string

GetPendingAuthServers returns the names of all servers awaiting authentication.

func (*SessionState) RemoveConnection ¶

func (s *SessionState) RemoveConnection(serverName string)

RemoveConnection removes the session's connection to a specific server. This is used when logging out from a server or when a connection becomes invalid.

func (*SessionState) SetConnection ¶

func (s *SessionState) SetConnection(serverName string, conn *SessionConnection)

SetConnection sets the session's connection to a specific server.

func (*SessionState) UpdateActivity ¶

func (s *SessionState) UpdateActivity()

UpdateActivity updates the last activity timestamp for the session.

func (*SessionState) UpgradeConnection ¶

func (s *SessionState) UpgradeConnection(serverName string, client MCPClient, tokenKey *oauth.TokenKey) error

UpgradeConnection upgrades a connection from pending_auth to connected.

This method validates the connection state before upgrading:

• The connection must exist
• The connection must be in pending_auth state (not already connected)

Returns an error if validation fails.