Python warning cleanup

src/lsf_mcp_server/__init__.py

@@ -20,4 +20,4 @@
 # when running with python -m lsf_mcp_server.server
 # Users should import directly: from lsf_mcp_server.server import LSFMCPServer, main
 
-__all__ = ['__version__']
+__all__ = ["__version__"]

src/lsf_mcp_server/auth.py

@@ -16,19 +16,19 @@
 
 import logging
 from typing import Dict, Optional
-from .lsf_client import LSFClient
 
+from .lsf_client import LSFClient
 
 logger = logging.getLogger(__name__)
 
 
 class AuthManager:
     """Manages authentication with LSF REST API."""
-    
+
     def __init__(self, client: LSFClient, username: str, password: str):
         """
         Initialize authentication manager.
-        
+
         Args:
             client: LSF API client
             username: LSF username
@@ -38,88 +38,88 @@ def __init__(self, client: LSFClient, username: str, password: str):
         self.username = username
         self.password = password
         self.session_info: Optional[Dict] = None
-        
+
     async def login(self) -> Dict:
         """
         Authenticate with LSF REST API.
-        
+
         Returns:
             Session information from the API
-            
+
         Raises:
             Exception: If authentication fails
         """
-        logger.info(f"Logging in as user: {self.username}")
-        
+        logger.info("Logging in as user: {self.username}")
+
         try:
             response = await self.client.post(
-                '/lsf/v1/auth/logon',
+                "/lsf/v1/auth/logon",
                 json={
-                    'name': self.username,
-                    'originalName': self.username,
-                    'pass': self.password
-                }
+                    "name": self.username,
+                    "originalName": self.username,
+                    "pass": self.password,
+                },
             )
-            
+
             session_data = response.json()
-            
+
             # Extract session token from response
             # The token is typically in the Set-Cookie header or response body
-            if 'token' in session_data:
-                token = session_data['token']
+            if "token" in session_data:
+                token = session_data["token"]
             else:
                 # Try to extract from cookies
                 cookies = response.cookies
-                if 'LSF_SESSION' in cookies:
-                    token = cookies['LSF_SESSION']
+                if "LSF_SESSION" in cookies:
+                    token = cookies["LSF_SESSION"]
                 else:
                     raise Exception("No session token found in response")
-            
+
             self.client.set_session_token(token)
             self.session_info = session_data
-            
+
             logger.info("Successfully authenticated with LSF API")
             return session_data
-            
+
         except Exception as e:
-            logger.error(f"Authentication failed: {str(e)}")
+            logger.error("Authentication failed: %s", str(e))
             raise Exception(f"Failed to authenticate with LSF API: {str(e)}")
-            
+
     async def logout(self):
         """
         Log out from LSF REST API.
-        
+
         Raises:
             Exception: If logout fails
         """
         if not self.session_info:
             logger.warning("No active session to logout")
             return
-            
+
         logger.info("Logging out from LSF API")
-        
+
         try:
-            await self.client.post('/lsf/v1/auth/logout')
+            await self.client.post("/lsf/v1/auth/logout")
             self.client.clear_session_token()
             self.session_info = None
             logger.info("Successfully logged out")
-            
+
         except Exception as e:
-            logger.error(f"Logout failed: {str(e)}")
+            logger.error("Logout failed: %s", str(e))
             # Clear session anyway
             self.client.clear_session_token()
             self.session_info = None
-            
+
     async def ensure_authenticated(self):
         """
         Ensure we have a valid session, re-authenticate if needed.
-        
+
         This method can be called before making API requests to ensure
         the session is still valid.
         """
         if not self.session_info:
             await self.login()
-            
+
     def is_authenticated(self) -> bool:
         """Check if currently authenticated."""
-        return self.session_info is not None
+        return self.session_info is not None

src/lsf_mcp_server/lsf_client.py

@@ -16,144 +16,138 @@
 
 import base64
 import logging
-from typing import Any, Dict, Optional
-import httpx
+from typing import Dict, Optional
 
+import httpx
 
 logger = logging.getLogger(__name__)
 
 
 class LSFClient:
     """HTTP client for LSF REST API."""
-    
+
     def __init__(self, base_url: str, timeout: float = 30.0):
         """
         Initialize LSF API client.
-        
+
         Args:
             base_url: Base URL of the LSF REST API (e.g., http://host:8088)
             timeout: Request timeout in seconds
         """
-        self.base_url = base_url.rstrip('/')
+        self.base_url = base_url.rstrip("/")
         self.timeout = timeout
         self.session_token: Optional[str] = None
         self._client = httpx.AsyncClient(timeout=timeout)
-        
+
     async def close(self):
         """Close the HTTP client."""
         await self._client.aclose()
-        
+
     def set_session_token(self, token: str):
         """Set the session token for authenticated requests."""
         self.session_token = token
-        
+
     def clear_session_token(self):
         """Clear the session token."""
         self.session_token = None
-
-    def _get_headers(self, additional_headers: Optional[Dict[str, str]] = None) -> Dict[str, str]:
+
+    def _get_headers(
+        self, additional_headers: Optional[Dict[str, str]] = None
+    ) -> Dict[str, str]:
         """
         Get headers for requests, including session token if available.
-        
+
         Args:
             additional_headers: Additional headers to include
-            
+
         Returns:
             Dictionary of headers
         """
         headers = {}
-        
+
         if self.session_token:
-            headers['Authorization'] = self.session_token
-            
+            headers["Authorization"] = self.session_token
+
         if additional_headers:
             headers.update(additional_headers)
-            
+
         return headers
-
-    async def request(
-        self,
-        method: str,
-        endpoint: str,
-        **kwargs
-    ) -> httpx.Response:
+
+    async def request(self, method: str, endpoint: str, **kwargs) -> httpx.Response:
         """
         Make an HTTP request to the LSF API.
-        
+
         Args:
             method: HTTP method (GET, POST, DELETE, etc.)
             endpoint: API endpoint (e.g., /v1/cluster)
             **kwargs: Additional arguments to pass to httpx
-            
+
         Returns:
             HTTP response
-            
+
         Raises:
             httpx.HTTPError: If the request fails
         """
         url = f"{self.base_url}{endpoint}"
-        
+
         # Merge headers
-        headers = self._get_headers(kwargs.pop('headers', None))
-        
-        logger.debug(f"{method} {url}")
-        
+        headers = self._get_headers(kwargs.pop("headers", None))
+
+        logger.debug("%s %s", method, url)
+
         try:
             response = await self._client.request(
-                method=method,
-                url=url,
-                headers=headers,
-                **kwargs
+                method=method, url=url, headers=headers, **kwargs
             )
-            
-            logger.debug(f"Response status: {response.status_code}")
-            
+
+            logger.debug("Response status: %s", response.status_code)
+
             # Raise for 4xx and 5xx status codes
             response.raise_for_status()
-            
+
             return response
-            
+
         except httpx.HTTPStatusError as e:
-            logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}")
+            logger.error("HTTP error: %s - %s", e.response.status_code, e.response.text)
             raise
         except httpx.RequestError as e:
-            logger.error(f"Request error: {str(e)}")
+            logger.error("Request error: %s", str(e))
             raise
-            
+
     async def get(self, endpoint: str, **kwargs) -> httpx.Response:
         """Make a GET request."""
-        return await self.request('GET', endpoint, **kwargs)
-        
+        return await self.request("GET", endpoint, **kwargs)
+
     async def post(self, endpoint: str, **kwargs) -> httpx.Response:
         """Make a POST request."""
-        return await self.request('POST', endpoint, **kwargs)
-        
+        return await self.request("POST", endpoint, **kwargs)
+
     async def delete(self, endpoint: str, **kwargs) -> httpx.Response:
         """Make a DELETE request."""
-        return await self.request('DELETE', endpoint, **kwargs)
-        
+        return await self.request("DELETE", endpoint, **kwargs)
+
     @staticmethod
     def encode_path(path: str) -> str:
         """
         Encode a file path to base64 for use in API endpoints.
-        
+
         Args:
             path: File path to encode
-            
+
         Returns:
             Base64 encoded path
         """
         return base64.b64encode(path.encode()).decode()
-        
+
     @staticmethod
     def decode_path(encoded_path: str) -> str:
         """
         Decode a base64 encoded file path.
-        
+
         Args:
             encoded_path: Base64 encoded path
-            
+
         Returns:
             Decoded file path
         """
-        return base64.b64decode(encoded_path.encode()).decode()
+        return base64.b64decode(encoded_path.encode()).decode()