"""
Session-scoped CKAN configuration state management.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from mcp.server import Server
from .types import CkanToolsConfig
class SessionConfigError(Exception):
"""Base exception for session configuration failures."""
class MissingSessionConfigError(SessionConfigError):
"""Raised when a CKAN configuration has not been initialised."""
def __init__(self) -> None:
super().__init__(
"No CKAN API endpoint is initialised for this session. "
"Call 'ckan_api_initialise' to select a portal."
)
@dataclass
class SessionState:
"""Session-specific state container."""
config: CkanToolsConfig | None = None
metadata: dict[str, Any] = field(default_factory=dict)
class SessionConfigStore:
"""
Store CKAN configuration state for each connected MCP session.
The MCP server provides a new session object for each client connection,
so using that session object's identity gives us a per-client key without
resorting to global singletons.
"""
def __init__(self, server: Server):
self._server = server
self._states: dict[int, SessionState] = {}
def _current_key(self) -> int:
"""Return the dictionary key representing the active session."""
request_context = self._server.request_context
return id(request_context.session)
def _current_state(self) -> SessionState:
"""Return the state for the active session, creating it if needed."""
key = self._current_key()
if key not in self._states:
self._states[key] = SessionState()
return self._states[key]
def has_config(self) -> bool:
"""Return True when the active session has an initialised config."""
return self._current_state().config is not None
def get_config(self) -> CkanToolsConfig | None:
"""Return the active session's CKAN configuration, if any."""
return self._current_state().config
def require_config(self) -> CkanToolsConfig:
"""Return the active session's CKAN configuration, raising if missing."""
config = self.get_config()
if config is None:
raise MissingSessionConfigError()
return config
def set_config(
self, config: CkanToolsConfig, *, metadata: dict[str, Any] | None = None
) -> None:
"""Persist a CKAN configuration for the current session."""
state = self._current_state()
state.config = config
state.metadata = metadata or {}
def clear(self) -> None:
"""Remove any stored CKAN configuration for the current session."""
state = self._current_state()
state.config = None
state.metadata = {}
def get_metadata(self) -> dict[str, Any]:
"""Return metadata about the selected CKAN endpoint for the session."""
return dict(self._current_state().metadata)
