Pylint improvements (IBM#332)

* Pylint cli

* Pylint sampling

* Pylint schemas

* Pyling completion_services

* Pylint resource_service

* Pylint streamablehttp_transport

* Pylint websocket_transport

* Pylint mcpgateway/admin.py

* Pylint mcpgateway/federation/discovery.py

* Pylint mcpgateway/federation/forward.py

* Pylint mcpgateway/federation/manager.py

* Pylint mcpgateway/services/prompt_service.py

* Pylint mcpgateway/services/server_service.py

* Pylint mcpgateway/services/tool_service.py

* Pylint mcpgateway/transports/sse_transport.py

* Pylint mcpgateway/version.py

* Pylint mcpgateway/wrapper.py

* Remove unneccessary pylint ignore comments

* Load pylint-pydantic plugin in .pylintrc

* Pylint mcpgateway/db.py

* Pylint mcpgateway/main.py

* Pylint mcpgateway/translate.py

* Pylint mcpgateway/cache/session_registry.py

* Pylint mcpgateway/translate.py

Signed-off-by: Sebastian <sebastian.iozu@ibm.com>

* Pylint part of mcpgateway/services/gateway_service.py

Signed-off-by: Sebastian <sebastian.iozu@ibm.com>

* Fix typo in mcpgateway/translate.py

Signed-off-by: Sebastian <sebastian.iozu@ibm.com>

* Merge main and small fixes

Signed-off-by: Sebastian <sebastian.iozu@ibm.com>

* Fix typo in mcpgateway/cache/session_registry.py

Signed-off-by: Sebastian <sebastian.iozu@ibm.com>

* Fix flake8

Signed-off-by: Sebastian <sebastian.iozu@ibm.com>

---------

Signed-off-by: Sebastian <sebastian.iozu@ibm.com>
Co-authored-by: Sebastian <sebastian.iozu@ibm.com>

Author: ChrisPC-39

mcpgateway/bootstrap_db.py

@@ -25,11 +25,11 @@
 import logging
 
 # Third-Party
-from alembic.config import Config
 from sqlalchemy import create_engine, inspect
 
 # First-Party
 from alembic import command
+from alembic.config import Config
 from mcpgateway.config import settings
 from mcpgateway.db import Base
 

mcpgateway/cache/session_registry.py

@@ -570,7 +570,7 @@ def _db_cleanup():
                             continue
 
                         # Refresh session in database
-                        def _refresh_session():
+                        def _refresh_session(session_id=session_id):
                             db_session = next(get_db())
                             try:
                                 session = db_session.query(SessionRecord).filter(SessionRecord.session_id == session_id).first()

mcpgateway/handlers/sampling.py

@@ -19,6 +19,7 @@
 # First-Party
 from mcpgateway.models import CreateMessageResult, ModelPreferences, Role, TextContent
 
+
 logger = logging.getLogger(__name__)
 
 

mcpgateway/main.py

@@ -290,14 +290,14 @@ class MCPPathRewriteMiddleware:
     - All other requests are passed through without change.
     """
 
-    def __init__(self, app):
+    def __init__(self, application):
         """
         Initialize the middleware with the ASGI application.
 
         Args:
-            app (Callable): The next ASGI application in the middleware stack.
+            application (Callable): The next ASGI application in the middleware stack.
         """
-        self.app = app
+        self.application = application
 
     async def __call__(self, scope, receive, send):
         """
@@ -310,7 +310,7 @@ async def __call__(self, scope, receive, send):
         """
         # Only handle HTTP requests, HTTPS uses scope["type"] == "http" in ASGI
         if scope["type"] != "http":
-            await self.app(scope, receive, send)
+            await self.application(scope, receive, send)
             return
 
         # Call auth check first
@@ -325,7 +325,7 @@ async def __call__(self, scope, receive, send):
             scope["path"] = "/mcp"
             await streamable_http_session.handle_streamable_http(scope, receive, send)
             return
-        await self.app(scope, receive, send)
+        await self.application(scope, receive, send)
 
 
 # Configure CORS

mcpgateway/services/gateway_service.py

@@ -32,6 +32,15 @@
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
+try:
+    # Third-Party
+    import redis
+
+    REDIS_AVAILABLE = True
+except ImportError:
+    REDIS_AVAILABLE = False
+    logging.info("Redis is not utilized in this environment.")
+
 # First-Party
 from mcpgateway.config import settings
 from mcpgateway.db import Gateway as DbGateway
@@ -42,15 +51,6 @@
 from mcpgateway.utils.create_slug import slugify
 from mcpgateway.utils.services_auth import decode_auth
 
-try:
-    # Third-Party
-    import redis
-
-    REDIS_AVAILABLE = True
-except ImportError:
-    REDIS_AVAILABLE = False
-    logging.info("Redis is not utilized in this environment.")
-
 # logging.getLogger("httpx").setLevel(logging.WARNING)  # Disables httpx logs for regular health checks
 logger = logging.getLogger(__name__)
 
@@ -247,16 +247,16 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
 
             return GatewayRead.model_validate(gateway)
         except* GatewayConnectionError as ge:
-            logger.error("GatewayConnectionError in group: %s", ge.exceptions)
+            logger.error(f"GatewayConnectionError in group: {ge.exceptions}")
             raise ge.exceptions[0]
         except* ValueError as ve:
-            logger.error("ValueErrors in group: %s", ve.exceptions)
+            logger.error(f"ValueErrors in group: {ve.exceptions}")
             raise ve.exceptions[0]
         except* RuntimeError as re:
-            logger.error("RuntimeErrors in group: %s", re.exceptions)
+            logger.error(f"RuntimeErrors in group: {re.exceptions}")
             raise re.exceptions[0]
         except* BaseException as other:  # catches every other sub-exception
-            logger.error("Other grouped errors: %s", other.exceptions)
+            logger.error(f"Other grouped errors: {other.exceptions}")
             raise other.exceptions[0]
 
     async def list_gateways(self, db: Session, include_inactive: bool = False) -> List[GatewayRead]:

mcpgateway/services/prompt_service.py

@@ -30,8 +30,8 @@
 # First-Party
 from mcpgateway.db import Prompt as DbPrompt
 from mcpgateway.db import PromptMetric, server_prompt_association
-from mcpgateway.models import Message, PromptResult, Role, TextContent
 from mcpgateway.schemas import PromptCreate, PromptRead, PromptUpdate
+from mcpgateway.models import Message, PromptResult, Role, TextContent
 
 logger = logging.getLogger(__name__)
 

mcpgateway/translate.py

@@ -55,11 +55,11 @@
 import uuid
 
 # Third-Party
+import uvicorn
+from sse_starlette.sse import EventSourceResponse
 from fastapi import FastAPI, Request, Response, status
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import PlainTextResponse
-from sse_starlette.sse import EventSourceResponse
-import uvicorn
 
 try:
     # Third-Party
@@ -135,7 +135,7 @@ async def start(self) -> None:
 
         Creates the subprocess and starts the stdout pump task.
         """
-        LOGGER.info("Starting stdio subprocess: %s", self._cmd)
+        LOGGER.info(f"Starting stdio subprocess: {self._cmd}")
         self._proc = await asyncio.create_subprocess_exec(
             *shlex.split(self._cmd),
             stdin=asyncio.subprocess.PIPE,
@@ -153,7 +153,7 @@ async def stop(self) -> None:
         """
         if self._proc is None:
             return
-        LOGGER.info("Stopping subprocess (pid=%s)", self._proc.pid)
+        LOGGER.info(f"Stopping subprocess (pid={self._proc.pid})")
         self._proc.terminate()
         with suppress(asyncio.TimeoutError):
             await asyncio.wait_for(self._proc.wait(), timeout=5)
@@ -171,7 +171,7 @@ async def send(self, raw: str) -> None:
         """
         if not self._stdin:
             raise RuntimeError("stdio endpoint not started")
-        LOGGER.debug("→ stdio: %s", raw.strip())
+        LOGGER.debug(f"→ stdio: {raw.strip()}")
         self._stdin.write(raw.encode())
         await self._stdin.drain()
 
@@ -192,7 +192,7 @@ async def _pump_stdout(self) -> None:
                 if not line:  # EOF
                     break
                 text = line.decode(errors="replace")
-                LOGGER.debug("← stdio: %s", text.strip())
+                LOGGER.debug(f"← stdio: {text.strip()}")
                 await self._pubsub.publish(text)
         except Exception:  # pragma: no cover --best-effort logging
             LOGGER.exception("stdout pump crashed - terminating bridge")
@@ -420,7 +420,7 @@ async def _shutdown() -> None:
         with suppress(NotImplementedError):  # Windows lacks add_signal_handler
             loop.add_signal_handler(sig, lambda: asyncio.create_task(_shutdown()))
 
-    LOGGER.info("Bridge ready → http://127.0.0.1:%s/sse", port)
+    LOGGER.info(f"Bridge ready → http://127.0.0.1:{port}/sse")
     await server.serve()
     await _shutdown()  # final cleanup
 

mcpgateway/transports/streamablehttp_transport.py

@@ -50,6 +50,7 @@
 from mcpgateway.services.tool_service import ToolService
 from mcpgateway.utils.verify_credentials import verify_credentials
 
+
 logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.INFO)