MCP Variance Log

Instructions

            Schema Reference:
            Table: chat_monitoring
            Fields:
            - log_id (INTEGER PRIMARY KEY)
            - timestamp (DATETIME)
            - session_id (TEXT)
            - user_id (TEXT)
            - interaction_type (TEXT)
            - probability_class (TEXT: HIGH, MEDIUM, LOW)
            - message_content (TEXT)
            - response_content (TEXT)
            - context_summary (TEXT)
            - reasoning (TEXT)

            Example:
            SELECT timestamp, probability_class, context_summary 
            FROM chat_monitoring 
            WHERE probability_class = 'LOW'
            LIMIT 5;
            

Implementation Reference

• src/mcp_variance_log/server.py:268-275 (handler)

  Handler logic for the 'read_query' tool within the @server.call_tool() dispatcher. Validates the presence and SELECT prefix of the query argument, executes it via the database helper, and returns the results as TextContent.

  elif name == "read_query":
      if not arguments or "query" not in arguments:
          raise ValueError("Missing query argument")
      query = arguments["query"].strip()
      if not query.upper().startswith("SELECT"):
          raise ValueError("Only SELECT queries are allowed")
      results = db._execute_query(query)
      return [types.TextContent(type="text", text=str(results))]

• src/mcp_variance_log/server.py:173-182 (schema)

  JSON schema defining the input for the 'read_query' tool: an object with a required 'query' string property.

  inputSchema={
      "type": "object",
      "properties": {
          "query": {
              "type": "string",
              "description": "SELECT SQL query to execute"
          }
      },
      "required": ["query"]
  }

• src/mcp_variance_log/server.py:149-183 (registration)

  Registration of the 'read_query' tool returned by the @server.list_tools() handler, including name, detailed description with schema reference, and input schema.

  types.Tool(
      name="read_query",
      description="""Execute a SELECT query on the SQLite database
  
          Schema Reference:
          Table: chat_monitoring
          Fields:
          - log_id (INTEGER PRIMARY KEY)
          - timestamp (DATETIME)
          - session_id (TEXT)
          - user_id (TEXT)
          - interaction_type (TEXT)
          - probability_class (TEXT: HIGH, MEDIUM, LOW)
          - message_content (TEXT)
          - response_content (TEXT)
          - context_summary (TEXT)
          - reasoning (TEXT)
  
          Example:
          SELECT timestamp, probability_class, context_summary 
          FROM chat_monitoring 
          WHERE probability_class = 'LOW'
          LIMIT 5;
          """,
      inputSchema={
          "type": "object",
          "properties": {
              "query": {
                  "type": "string",
                  "description": "SELECT SQL query to execute"
              }
          },
          "required": ["query"]
      }
  ),

• src/mcp_variance_log/db_utils.py:29-52 (helper)

  Private helper method of LogDatabase class that executes arbitrary SQL queries (read or write), handles parameters, commits writes, and returns results as list of dictionaries. Used by read_query and other tools.

  def _execute_query(self, query: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
      """Execute a SQL query and return results as a list of dictionaries"""
      logger.debug(f"Executing query: {query}")
      try:
          with closing(sqlite3.connect(self.db_path)) as conn:
              conn.row_factory = sqlite3.Row
              with closing(conn.cursor()) as cursor:
                  if params:
                      cursor.execute(query, params)
                  else:
                      cursor.execute(query)
  
                  if query.strip().upper().startswith(('INSERT', 'UPDATE', 'DELETE', 'CREATE', 'DROP', 'ALTER')):
                      conn.commit()
                      affected = cursor.rowcount
                      logger.debug(f"Write query affected {affected} rows")
                      return [{"affected_rows": affected}]
  
                  results = [dict(row) for row in cursor.fetchall()]
                  logger.debug(f"Read query returned {len(results)} rows")
                  return results
      except Exception as e:
          logger.error(f"Database error executing query: {e}")
          raise