feat(clp-mcp-server): Introduce session-based MCP capabilities:

components/clp-mcp-server/README.md

@@ -49,6 +49,14 @@ For more details on Claude Desktop MCP setup, see the
 
 > **Warning:** 🚧 This section is still under construction.
 
+## Testing
+
+Use the following command to run all unit tests:
+
+```shell
+uv test pytest
+```
+
 [claude-desktop]: https://claude.ai/download
 [claude-desktop-mcp-doc]: https://modelcontextprotocol.io/docs/develop/connect-local-servers
 [mcp]: https://modelcontextprotocol.io/docs/getting-started/intro

components/clp-mcp-server/clp_mcp_server/server/constants.py

@@ -0,0 +1,26 @@
+"""Constants for CLP MCP server."""
+
+EXPIRED_SESSION_SWEEP_INTERVAL_SECONDS = 600  # 10 minutes
+NUM_ITEMS_PER_PAGE = 10
+MAX_CACHED_RESULTS = 1000
+SESSION_TTL_MINUTES = 60
+
+SERVER_NAME = "clp-mcp-server"
+SYSTEM_PROMPT = """
+You are an AI assistant that helps users query a log database using KQL (Kibana Query Language).
+ When given a user query, you should generate a KQL query that accurately captures the user's
+ intent. The KQL query should be as specific as possible to minimize the number of log messages
+ returned.
+
+You should also consider the following guidelines when generating KQL queries:
+- Use specific field names and values to narrow down the search.
+- Avoid using wildcards (*) unless absolutely necessary, as they can lead to large result
+ sets.
+- Use logical operators (AND, OR, NOT) to combine multiple conditions.
+- Consider the time range of the logs you are searching. If the user specifies a time range,
+ include it in the KQL query.
+- If the user query is ambiguous or lacks detail, ask clarifying questions to better
+ understand their intent before generating the KQL query.
+- Always ensure that the generated KQL query is syntactically correct and can be executed
+ without errors.
+"""

components/clp-mcp-server/clp_mcp_server/server/server.py

@@ -2,25 +2,10 @@
 
 from typing import Any
 
-from fastmcp import FastMCP
+from fastmcp import Context, FastMCP
 
-
-class ProtocolConstant:
-    """Constants for the CLP MCP Server."""
-
-    SERVER_NAME = "clp-mcp-server"
-
-    # Tool names
-    TOOL_HELLO_WORLD = "hello_world"
-    TOOL_GET_SERVER_INFO = "get_server_info"
-
-    @classmethod
-    def get_capabilities(cls) -> list[str]:
-        """
-        Gets the capabilities of the server.
-        :return: A list of tool names supported by the server.
-        """
-        return [cls.TOOL_HELLO_WORLD, cls.TOOL_GET_SERVER_INFO]
+from . import constants
+from .session_manager import SessionManager
 
 
 def create_mcp_server() -> FastMCP:
@@ -31,22 +16,44 @@ def create_mcp_server() -> FastMCP:
     :raise: Propagates `FastMCP.__init__`'s exceptions.
     :raise: Propagates `FastMCP.tool`'s exceptions.
     """
-    mcp = FastMCP(name=ProtocolConstant.SERVER_NAME)
+    mcp = FastMCP(name=constants.SERVER_NAME)
+
+    session_manager = SessionManager(session_ttl_minutes=constants.SESSION_TTL_MINUTES)
 
-    @mcp.tool()
-    def get_server_info() -> dict[str, Any]:
+    @mcp.tool
+    async def get_instructions(ctx: Context) -> str:
         """
-        Gets the MCP server's information.
+        Gets a pre-defined "system prompt" that guides the LLM behavior.
+        This function must be invoked before any other `FastMCP.tool`.
 
-        :return: The server's information with a list of capabilities.
+        :param ctx: The `FastMCP` context containing the metadata of the underlying MCP session.
+        :return: A string of "system prompt".
         """
-        return {
-            "name": ProtocolConstant.SERVER_NAME,
-            "capabilities": ProtocolConstant.get_capabilities(),
-            "status": "running",
-        }
+        await session_manager.start()
+        return session_manager.get_or_create_session(ctx.session_id).get_instructions()
+
+    @mcp.tool
+    async def get_nth_page(page_index: int, ctx: Context) -> dict[str, Any]:
+        """
+        Retrieves the n-th page of a paginated response with the paging metadata from the previous
+        query.
+
+        :param page_index: Zero-based index, e.g., 0 for the first page.
+        :param ctx: The `FastMCP` context containing the metadata of the underlying MCP session.
+        :return: A dictionary containing the following key-value pairs on success:
+            - "items": A list of log entries in the requested page.
+            - "total_pages": Total number of pages available from the query as an integer.
+            - "total_items": Total number of log entries available from the query as an integer.
+            - "num_items_per_page": Number of log entries per page.
+            - "has_next": Whether a page exists after the returned one.
+            - "has_previous": Whether a page exists before the returned one.
+        :return: A dictionary with the following key-value pair on failures:
+            - "Error": An error message describing the failure.
+        """
+        await session_manager.start()
+        return session_manager.get_nth_page(ctx.session_id, page_index)
 
-    @mcp.tool()
+    @mcp.tool
     def hello_world(name: str = "clp-mcp-server user") -> dict[str, Any]:
         """
         Provides a simple hello world greeting.
@@ -56,7 +63,7 @@ def hello_world(name: str = "clp-mcp-server user") -> dict[str, Any]:
         """
         return {
             "message": f"Hello World, {name.strip()}!",
-            "server": ProtocolConstant.SERVER_NAME,
+            "server": constants.SERVER_NAME,
             "status": "running",
         }
 

components/clp-mcp-server/clp_mcp_server/server/session_manager.py

@@ -0,0 +1,242 @@
+"""Session management for CLP MCP Server."""
+
+import asyncio
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any, ClassVar
+
+from paginate import Page
+
+from . import constants
+
+
+class PaginatedQueryResult:
+    """Paginates the cached log entries returned from a query's response."""
+
+    def __init__(self, result_log_entries: list[str], num_items_per_page: int) -> None:
+        """
+        :param result_log_entries: List of cached log entries to paginate.
+        :param num_items_per_page:
+        :raise: ValueError if the number of cached results or `num_items_per_page` is invalid.
+        """
+        if len(result_log_entries) > constants.MAX_CACHED_RESULTS:
+            err_msg = (
+                "PaginatedQueryResult exceeds maximum allowed cached results:"
+                f" {len(result_log_entries)} > {constants.MAX_CACHED_RESULTS}."
+            )
+            raise ValueError(err_msg)
+
+        if num_items_per_page <= 0:
+            err_msg = (
+                f"Invalid num_items_per_page: {num_items_per_page}, it must be a positive integer."
+            )
+            raise ValueError(err_msg)
+
+        self._num_items_per_page = num_items_per_page
+
+        self._num_pages = (len(result_log_entries) + num_items_per_page - 1) // num_items_per_page
+        self._result_log_entries = result_log_entries
+
+    def get_page(self, page_index: int) -> Page | None:
+        """
+        :param page_index: Zero-based index, e.g., 0 for the first page.
+        :return: A `Page` object for the specified page.
+        :return: None if `page_index` is out of bounds.
+        """
+        # Convert zero-based to one-based
+        page_number = page_index + 1
+        if page_number <= 0 or self._num_pages < page_number:
+            return None
+
+        return Page(
+            self._result_log_entries,
+            page=page_number,
+            items_per_page=self._num_items_per_page,
+        )
+
+
+@dataclass
+class SessionState:
+    """
+    Represents the state of a user session.
+    `SessionState` respects `FastMCP` asynchronous concurrency model that is built on Python's
+    asyncio runtime:
+    Asyncio achieves concurrency on a single thread by allowing tasks to yield control at `await`
+    points. This means when a `mcp.tool` API calls executes an await expression, it gets suspended,
+    and the event loop executes the next `mcp.tool` API calls from received requests issued by
+    clients.
+    Therefore, `SessionState` does not need thread-safe operations because a session cannot be
+    accessed from two threads at the same time.
+    """
+
+    _num_items_per_page: int
+    _session_id: str
+    _session_ttl_minutes: int
+
+    _cached_query_result: PaginatedQueryResult | None = None
+    _last_accessed: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    _is_instructions_retrieved: bool = False
+
+    _GET_INSTRUCTIONS_NOT_RUN_ERROR: ClassVar[dict[str, str]] = {
+        "Error": "Please call `get_instructions()` first to understand how to use this MCP server."
+    }
+
+    def cache_query_result_and_get_first_page(
+        self,
+        results: list[str],
+    ) -> dict[str, Any]:
+        """
+        :param results: Log entries from the query to cache.
+        :return: Forwards `SessionState.get_page_data`'s return values.
+        :return: _GET_INSTRUCTIONS_NOT_RUN_ERROR if `get_instructions` has not been called in this
+            session.
+        """
+        if self._is_instructions_retrieved is False:
+            return self._GET_INSTRUCTIONS_NOT_RUN_ERROR.copy()
+
+        self._cached_query_result = PaginatedQueryResult(
+            result_log_entries=results, num_items_per_page=self._num_items_per_page
+        )
+
+        return self.get_page_data(0)
+
+    def get_instructions(self) -> str:
+        """
+        Gets a pre-defined "system prompt" that guides the LLM behavior.
+
+        :return: A string of "system prompt".
+        """
+        self._is_instructions_retrieved = True
+        return constants.SYSTEM_PROMPT
+
+    def get_page_data(self, page_index: int) -> dict[str, Any]:
+        """
+        Retrieves the n-th page of a paginated response with the paging metadata from the previous
+        query.
+
+        NOTE: This docstring must be synchronized with `get_nth_page`'s MCP tool call.
+
+        :param page_index: Zero-based index, e.g., 0 for the first page.
+        :return: A dictionary containing the following key-value pairs on success:
+            - "items": A list of log entries in the requested page.
+            - "total_pages": Total number of pages available from the query as an integer.
+            - "total_items": Total number of log entries available from the query as an integer.
+            - "num_items_per_page": Number of log entries per page.
+            - "has_next": Whether a page exists after the returned one.
+            - "has_previous": Whether a page exists before the returned one.
+        :return: A dictionary with the following key-value pair on failures:
+            - "Error": An error message describing the failure.
+        :return: _GET_INSTRUCTIONS_NOT_RUN_ERROR if `get_instructions` has not been called in this
+            session.
+        """
+        if self._is_instructions_retrieved is False:
+            return self._GET_INSTRUCTIONS_NOT_RUN_ERROR.copy()
+
+        if self._cached_query_result is None:
+            return {"Error": "No previous paginated response in this session."}
+
+        page = self._cached_query_result.get_page(page_index)
+        if page is None:
+            return {"Error": "Page index is out of bounds."}
+
+        return {
+            "items": list(page),
+            "total_pages": page.page_count,
+            "total_items": page.item_count,
+            "num_items_per_page": page.items_per_page,
+            "has_next": page.next_page is not None,
+            "has_previous": page.previous_page is not None,
+        }
+
+    def is_expired(self) -> bool:
+        """:return: Whether the session has expired."""
+        time_diff = datetime.now(timezone.utc) - self._last_accessed
+        return time_diff > timedelta(minutes=self._session_ttl_minutes)
+
+    def update_access_time(self) -> None:
+        """Updates the last accessed timestamp."""
+        self._last_accessed = datetime.now(timezone.utc)
+
+
+class SessionManager:
+    """
+    Session manager for concurrent user sessions.
+    `SessionManager` respects `FastMCP` asynchronous concurrency model that is built on Python's
+    asyncio runtime:
+    Asyncio achieves concurrency on a single thread by allowing tasks to yield control at `await`
+    points. This means when a `mcp.tool` API calls executes an await expression, it gets suspended,
+    and the event loop executes the next `mcp.tool` API calls from received requests issued by
+    clients.
+    The operations on `sessions` which maintains concurrent user sessions are performed by
+    asynchronous tasks. The above concurrency model guarantees that the operations of `sessions`
+    are atomic because it cannot be accessed from two threads at the same time.
+    """
+
+    def __init__(self, session_ttl_minutes: int) -> None:
+        """:param session_ttl_minutes: Session time-to-live in minutes."""
+        self.sessions: dict[str, SessionState] = {}
+        self._session_ttl_minutes = session_ttl_minutes
+        self._cleanup_task: asyncio.Task | None = None
+
+    async def start(self) -> None:
+        """Starts the asynchronous cleanup task."""
+        if self._cleanup_task is None:
+            self._cleanup_task = asyncio.create_task(self._cleanup_loop())
+
+    async def _cleanup_loop(self) -> None:
+        """Cleans up all expired sessions periodically as an async task."""
+        while True:
+            await asyncio.sleep(constants.EXPIRED_SESSION_SWEEP_INTERVAL_SECONDS)
+            self.cleanup_expired_sessions()
+
+    def cleanup_expired_sessions(self) -> None:
+        """Cleans up all expired sessions."""
+        expired_sessions = [sid for sid, session in self.sessions.items() if session.is_expired()]
+
+        for sid in expired_sessions:
+            del self.sessions[sid]
+
+    def cache_query_result_and_get_first_page(
+        self, session_id: str, query_results: list[str]
+    ) -> dict[str, Any]:
+        """
+        :param session_id:
+        :param query_results: Log entries from the query to cache.
+        :return: Forwards `SessionState.cache_query_result_and_get_first_page`'s return values.
+        """
+        session = self.get_or_create_session(session_id)
+
+        return session.cache_query_result_and_get_first_page(results=query_results)
+
+    def get_nth_page(self, session_id: str, page_index: int) -> dict[str, Any]:
+        """
+        Retrieves the n-th page of a paginated response with the paging metadata from the previous
+        query.
+
+        :param session_id:
+        :param page_index: Index of the page to retrieve (zero-based, e.g., 0 is the first page).
+        :return: Forwards `SessionState.get_page_data`'s return values.
+        """
+        session = self.get_or_create_session(session_id)
+
+        return session.get_page_data(page_index)
+
+    def get_or_create_session(self, session_id: str) -> SessionState:
+        """
+        Gets an existing session or creates a new one.
+
+        :param session_id:
+        :return: The `SessionState` object for the given `session_id`.
+        """
+        if session_id in self.sessions and self.sessions[session_id].is_expired():
+            del self.sessions[session_id]
+
+        if session_id not in self.sessions:
+            self.sessions[session_id] = SessionState(
+                constants.NUM_ITEMS_PER_PAGE, session_id, self._session_ttl_minutes
+            )
+
+        session = self.sessions[session_id]
+
+        session.update_access_time()
+        return session