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

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

@@ -0,0 +1,27 @@
+"""Constants for CLP MCP server."""
+
+
+class CLPMcpConstants:
+    """Constants for the CLP MCP Server."""
+
+    CLEAN_UP_SECONDS = 600  # 10 minutes
+    ITEM_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 .constants import CLPMcpConstants
+from .session_manager import SessionManager
 
 
 def create_mcp_server() -> FastMCP:
@@ -31,22 +16,36 @@ 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=CLPMcpConstants.SERVER_NAME)
+
+    session_manager = SessionManager(CLPMcpConstants.SESSION_TTL_MINUTES)
 
-    @mcp.tool()
-    def get_server_info() -> dict[str, Any]:
+    @mcp.tool
+    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",
-        }
+        session = session_manager.get_or_create_session(ctx.session_id)
+        session.ran_instructions = True
+        return CLPMcpConstants.SYSTEM_PROMPT
+
+    @mcp.tool
+    def get_nth_page(page_index: int, ctx: Context) -> dict[str, Any]:
+        """
+        Retrieves the n-th page of a paginated response 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: On success, dictionary containing paged log entries and pagination metadata.
+        On error, dictionary with ``{"Error": "error message describing the failure"}``.
+        """
+        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 +55,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": CLPMcpConstants.SERVER_NAME,
             "status": "running",
         }
 

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

@@ -0,0 +1,206 @@
+"""Session management for CLP MCP Server."""
+
+import threading
+import time
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from paginate import Page
+
+from .constants import CLPMcpConstants
+
+
+@dataclass(frozen=True)
+class QueryResult:
+    """Cached results from previous query's response."""
+
+    total_results: list[str]
+    items_per_page: int
+
+    _total_pages: int = field(init=False, repr=False)
+
+    def __post_init__(self) -> None:
+        """Validates that cached results don't exceed MAX_CACHED_RESULTS."""
+        if len(self.total_results) > CLPMcpConstants.MAX_CACHED_RESULTS:
+            err_msg = (
+                f"QueryResult exceeds maximum allowed cached results: "
+                f"{len(self.total_results)} > {CLPMcpConstants.MAX_CACHED_RESULTS}. "
+            )
+            raise ValueError(err_msg)
+
+        object.__setattr__(
+            self,
+            "_total_pages",
+            (len(self.total_results) + self.items_per_page - 1) // self.items_per_page,
+        )
+
+    def get_page(self, page_number: int) -> Page | None:
+        """
+        Gets a specific page from the cached response.
+
+        :param page_number: One-based indexing, e.g., 1 for the first page
+        :return: Page object or None if page number is out of bounds
+        """
+        if page_number > self._total_pages or page_number <= 0:
+            return None
+
+        return Page(
+            self.total_results,
+            page=page_number,
+            items_per_page=self.items_per_page,
+        )
+
+
+@dataclass
+class SessionState:
+    """State of a single user session."""
+
+    session_id: str
+    items_per_page: int
+    session_ttl_minutes: int
+    last_accessed: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+    cached_query_result: QueryResult | None = None
+    ran_instructions: bool = False
+
+    def cache_query_result(
+        self,
+        results: list[str],
+    ) -> None:
+        """
+        Caches the latest query result of the session.
+
+        :param query_results: Complete log entries to cache
+        """
+        self.cached_query_result = QueryResult(
+            total_results=results, items_per_page=self.items_per_page
+        )
+
+    def get_page_data(self, page_number: int) -> dict[str, Any]:
+        """
+        Gets page data in a dictionary format.
+
+        :param page_number: One-based indexing, e.g., 1 for the first page
+        :return: On success, dictionary containing paged log entries and pagination metadata.
+        On error, dictionary with ``{"Error": "error message describing the failure"}``.
+        """
+        if self.cached_query_result is None:
+            return {"Error": "No previous paginated response in this session."}
+
+        page = self.cached_query_result.get_page(page_number)
+        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,
+            "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 all user sessions."""
+
+    def __init__(self, session_ttl_minutes: int) -> None:
+        """
+        Initializes the SessionManager and starts background cleanup thread.
+
+        :param session_ttl_minutes: Session time-to-live in minutes.
+        """
+        self.session_ttl_minutes = session_ttl_minutes
+        # sessions is a shared variable as there may be multiple session attached to the MCP server
+        # session state is NOT a shared variable because each session is accessed by only one
+        # connection at a time, and API calls for a single session are synchronous.
+        self._sessions_lock = threading.Lock()
+        self.sessions: dict[str, SessionState] = {}
+        self._cleanup_thread = threading.Thread(target=self._cleanup_loop, daemon=True)
+        self._cleanup_thread.start()
+
+    def _cleanup_loop(self) -> None:
+        """Cleans up all expired sessions periodically in a separate cleanup thread."""
+        while True:
+            time.sleep(CLPMcpConstants.CLEAN_UP_SECONDS)
+            self.cleanup_expired_sessions()
+
+    def cleanup_expired_sessions(self) -> None:
+        """Cleans up all expired sessions."""
+        with self._sessions_lock:
+            expired_sessions = [
+                sid for sid, session in self.sessions.items() if session.is_expired()
+            ]
+
+            for sid in expired_sessions:
+                del self.sessions[sid]
+
+    def get_or_create_session(self, session_id: str) -> SessionState:
+        """
+        Gets an existing session or creates a new one.
+
+        :param session_id: Unique identifier for the session
+        :return: The SessionState object for the given session_id
+        """
+        with self._sessions_lock:
+            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(
+                    session_id, CLPMcpConstants.ITEM_PER_PAGE, self.session_ttl_minutes
+                )
+
+            session = self.sessions[session_id]
+
+            session.update_access_time()
+            return session
+
+    def cache_query_result(self, session_id: str, query_results: list[str]) -> dict[str, Any]:
+        """
+        Caches query results for a session and return the first page.
+
+        :param session_id: Unique identifier for the session
+        :param query_results: Complete log entries to cache
+        :return: On success, dictionary containing the first page of log entries and
+        pagination metadata. On error, dictionary with
+        ``{"Error": "error message describing the failure"}``.
+        """
+        session = self.get_or_create_session(session_id)
+        if session.ran_instructions is False:
+            return {
+                "Error": "Please call get_instructions() first"
+                "to understand how to use this MCP server."
+            }
+
+        session.cache_query_result(results=query_results)
+
+        return session.get_page_data(1)
+
+    def get_nth_page(self, session_id: str, page_index: int) -> dict[str, Any]:
+        """
+        Retrieves the n-th page of a paginated response from the previous query.
+
+        :param session_id: Unique identifier for the session
+        :param page_index: Zero-based index, e.g., 0 for the first page
+        :return: On success, dictionary containing paged log entries and pagination metadata.
+        On error, dictionary with ``{"Error": "error message describing the failure"}``.
+        """
+        session = self.get_or_create_session(session_id)
+        if session.ran_instructions is False:
+            return {
+                "Error": "Please call get_instructions() first"
+                "to understand how to use this MCP server."
+            }
+
+        page_number = page_index + 1  # Convert zero-based to one-based
+        return session.get_page_data(page_number)

components/clp-mcp-server/pyproject.toml

@@ -6,11 +6,12 @@ authors = [{name = "YScope Inc.", email = "dev@yscope.com"}]
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
+    "aiomysql>=0.2.0",
     "click>=8.3.0",
     "fastmcp>=2.12.4",
-    "pymongo>=4.15.1",
-    "aiomysql>=0.2.0",
     "msgpack>=1.1.1",
+    "paginate>=0.5.7",
+    "pymongo>=4.15.1",
 ]
 
 [project.scripts]
@@ -23,10 +24,11 @@ build-backend = "hatchling.build"
 [dependency-groups]
 dev = [
     "mypy>=1.16.0",
-    "ruff>=0.11.12",
     "pytest>=8.4.1",
-    "pytest-env>=1.1.5",
     "pytest-asyncio>=1.2.0",
+    "pytest-env>=1.1.5",
+    "pytest-repeat>=0.9.4",
+    "ruff>=0.11.12",
 ]
 
 [tool.hatch.metadata]
@@ -69,12 +71,12 @@ ignore = [
 ]
 isort.order-by-type = false
 
+[tool.ruff.lint.per-file-ignores]
+"tests/**" = [
+    "S101",  # Allow usage of pytest `assert`
+    "TC003",  # Ignore performance overhead of imports only used for type checking
+]
+
 [tool.ruff.format]
 docstring-code-format = true
 docstring-code-line-length = 100
-
-[tool.ruff.lint.per-file-ignores]
-    "tests/test_clp_connector.py" = [
-        "INP001",  # Allow implicit namespace package for tests
-        "S101",  # Allow "assert" in test cases
-    ]

components/clp-mcp-server/tests/__init__.py

@@ -0,0 +1 @@
+"""Test Package for CLP MCP Server."""