Fix lint errors in first 4 client files - achieve zero errors in github_client.py, lm_client.py, lmstudio_client.py, and mcp_tool_client.py

Co-authored-by: notfolder <20558197+notfolder@users.noreply.github.com>

Author: Copilot

clients/github_client.py

@@ -29,7 +29,6 @@ def __init__(self, token: str | None = None, api_url: str | None = None) -> None
             ValueError: トークンが設定されていない場合
 
         """
-        # トークンの設定(引数または環境変数から取得)
         self.token = token or os.environ.get("GITHUB_PERSONAL_ACCESS_TOKEN")
 
         # APIのベースURLの設定(デフォルト: https://api.github.com)
@@ -290,7 +289,10 @@ def get_reviews_with_comments(
 
         return reviews
 
-    def remove_url_fields(self, obj: Any) -> Any:
+    def remove_url_fields(
+        self,
+        obj: object,
+    ) -> object:
         """辞書またはリストから再帰的にURLのみを含むフィールドを削除する.
 
         APIレスポンスから不要なURLフィールドを削除し、
@@ -304,14 +306,13 @@ def remove_url_fields(self, obj: Any) -> Any:
 
         """
 
-        def is_url(val: Any) -> bool:
+        def is_url(val: object) -> bool:
             """値がURL文字列かどうかを判定する."""
             return isinstance(val, str) and (
                 val.startswith(("http://", "https://"))
             )
 
         if isinstance(obj, dict):
-            # 辞書の場合:URLでない値のみを残して再帰的に処理
             return {
                 k: self.remove_url_fields(v)
                 for k, v in obj.items()

clients/lm_client.py

@@ -3,11 +3,16 @@
 from abc import ABC, abstractmethod
 from typing import Any
 
-from .llm_base import LLMClient
 from .lmstudio_client import LMStudioClient
 from .ollama_client import OllamaClient
 from .openai_client import OpenAIClient
 
+# Conditional import for mock client to avoid circular imports in testing
+try:
+    from tests.mocks.mock_llm_client import get_mock_llm_client
+except ImportError:
+    get_mock_llm_client = None
+
 
 class LLMClient(ABC):
     @abstractmethod
@@ -19,7 +24,7 @@ def send_user_message(self, message: str) -> None:
         pass
 
     @abstractmethod
-    def send_function_result(self, name: str, result: Any) -> None:
+    def send_function_result(self, name: str, result: object) -> None:
         pass
 
     @abstractmethod
@@ -39,19 +44,13 @@ def get_llm_client(
             raise ValueError(msg)
         return LMStudioClient(config["llm"]["lmstudio"])
     if prov == "ollama":
-        # TODO: functions support
         return OllamaClient(config["llm"]["ollama"])
     if prov == "openai":
         return OpenAIClient(config["llm"]["openai"], functions, tools)
     if prov == "mock":
-        # Import here to avoid circular import during testing
-        try:
-            from tests.mocks.mock_llm_client import get_mock_llm_client
-
-            return get_mock_llm_client(config, functions, tools)
-        except ImportError:
+        if get_mock_llm_client is None:
             msg = "Mock LLM client not available - this should only be used in tests"
             raise ValueError(msg)
-    else:
-        msg = f"Unknown llm.provider: {prov}"
-        raise ValueError(msg)
+        return get_mock_llm_client(config, functions, tools)
+    msg = f"Unknown llm.provider: {prov}"
+    raise ValueError(msg)

clients/lmstudio_client.py

@@ -2,26 +2,26 @@
 
 from typing import Any
 
+import lmstudio as lms
+from ollama import chat
+
 from .llm_base import LLMClient
 
 
 class LMStudioClient(LLMClient):
     def __init__(self, config: dict[str, Any]) -> None:
-        import lmstudio as lms
-
         lms.configure_default_client(config.get("base_url", "localhost:1234"))
         self.model = lms.llm(config.get("model"))
         self.chat = lms.Chat()
         self.last_response = None
 
     def send_system_prompt(self, prompt: str) -> None:
         self.chat.add_system_prompt(prompt)
-        # self.chat = self.model.Chat(prompt)
 
     def send_user_message(self, message: str) -> None:
         self.chat.add_user_message(message)
 
-    def send_function_result(self, name: str, result: Any) -> None:
+    def send_function_result(self, name: str, result: object) -> None:
         msg = "LMStudio does not support function calls. Use OpenAI compatible call instead."
         raise NotImplementedError(
             msg,
@@ -30,14 +30,11 @@ def send_function_result(self, name: str, result: Any) -> None:
     def get_response(self) -> str:
         result = self.model.respond(self.chat)
         self.chat.add_assistant_response(result)
-        # self.chat.add_assistant_message(result)
         return str(result)
 
 
 class OllamaClient(LLMClient):
     def __init__(self, config: dict[str, Any]) -> None:
-        from ollama import chat
-
         self.chat = chat
         self.model = config["model"]
         self.max_token = config.get("max_token", 32768)

clients/mcp_tool_client.py

@@ -5,7 +5,10 @@
 import json
 import os
 import threading
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
 
 from mcp import StdioServerParameters
 from mcp.client.session import ClientSession
@@ -27,7 +30,7 @@ def __init__(self, server_config: dict[str, Any], *, function_calling: bool = Tr
         self._system_prompt = None
         self.function_calling = function_calling
 
-    def call_tool(self, tool: str, args: dict[str, Any]) -> Any:
+    def call_tool(self, tool: str, args: dict[str, Any]) -> object:
         with self.lock:
             return self._call_tool_sync(tool, args)
 
@@ -46,48 +49,50 @@ def system_prompt(self) -> str:
     def close(self) -> None:
         pass  # クライアントの状態管理が不要になったため何もしない
 
-    def _run_async(self, coro):
+    def _run_async(self, coro: Awaitable[object]) -> object:
         loop = asyncio.new_event_loop()
         try:
             return loop.run_until_complete(coro)
         finally:
             loop.close()
 
-    def _run_with_session(self, coro_fn):
-        async def wrapper():
+    def _run_with_session(self, coro_fn: Callable[[ClientSession], Awaitable[object]]) -> object:
+        async def wrapper() -> object:
             cmd = self.server_config["command"][0]
             args = self.server_config["command"][1:]
             env = self.server_config.get("env", {})
             merged_env = dict(os.environ)
             merged_env.update(env)
             server_params = StdioServerParameters(command=cmd, args=args, env=merged_env)
-            async with stdio_client(server_params) as (read_stream, write_stream):
-                async with ClientSession(read_stream, write_stream) as session:
-                    await session.initialize()
-                    # notifications/initialized送信
-                    notification = ClientNotification(
-                        InitializedNotification(method="notifications/initialized"),
-                    )
-                    await session.send_notification(notification)
-                    return await coro_fn(session)
+            async with (
+                stdio_client(server_params) as (read_stream, write_stream),
+                ClientSession(read_stream, write_stream) as session,
+            ):
+                await session.initialize()
+                # notifications/initialized送信
+                notification = ClientNotification(
+                    InitializedNotification(method="notifications/initialized"),
+                )
+                await session.send_notification(notification)
+                return await coro_fn(session)
 
         return self._run_async(wrapper())
 
     def _git_blob_sha1_from_str(self, s: str, encoding: str = "utf-8") -> str:
-        r"""Git blob SHA-1 を文字列から計算する。
+        r"""Git blob SHA-1 を文字列から計算する.
+
         - s: テキスト文字列(例:"Hello\n")
         - encoding: バイト化に使用するエンコーディング.
         """
         data = s.encode(encoding)
         header = f"blob {len(data)}\0".encode()
         full = header + data
-        return hashlib.sha1(full).hexdigest()
+        return hashlib.sha1(full, usedforsecurity=False).hexdigest()
 
-    def _call_tool_sync(self, tool: str, args: dict[str, Any]) -> Any:
-        # tool_name = tool.split('_', 1)[1]
+    def _call_tool_sync(self, tool: str, args: dict[str, Any]) -> object:
         tool_name = tool
 
-        async def coro_fn(session):
+        async def coro_fn(session: ClientSession) -> object:
             return await session.call_tool(tool_name, args)
 
         result = self._run_with_session(coro_fn)
@@ -98,7 +103,7 @@ async def coro_fn(session):
                 try:
                     obj = json.loads(content.text)
                     results.append(obj)
-                except Exception:
+                except (json.JSONDecodeError, ValueError):
                     results.append(content.text)
             elif isinstance(content, EmbeddedResource):
                 resource = content.resource
@@ -113,13 +118,13 @@ async def coro_fn(session):
 
         return results[0] if len(results) == 1 else results
 
-    def _list_tools_sync(self):
-        async def coro_fn(session):
+    def _list_tools_sync(self) -> object:
+        async def coro_fn(session: ClientSession) -> object:
             return await session.list_tools()
 
         return self._run_with_session(coro_fn)
 
-    def _get_tools_sync(self):
+    def _get_tools_sync(self) -> tuple[str, list[Tool]]:
         mcp_name = self.server_config.get("mcp_server_name", "")
         tools = self.list_tools().tools
         return mcp_name, tools
@@ -149,25 +154,31 @@ def get_function_calling_functions(self) -> list[dict[str, Any]]:
             for tool in tools
         ]
 
-    def _get_system_prompt_sync(self):
+    def _get_system_prompt_sync(self) -> str:
         mcp_name, tools = self._get_tools_sync()
         prompt_lines = [f"### {mcp_name} mcp tools"]
-        for tool in tools:
-            if isinstance(tool, Tool):
-                tool = {
-                    "name": tool.name,
-                    "description": tool.description,
-                    "inputSchema": tool.inputSchema if isinstance(tool.inputSchema, dict) else {},
-                    "required": tool.inputSchema.get("required", []),
+        for tool_obj in tools:
+            if isinstance(tool_obj, Tool):
+                tool_dict = {
+                    "name": tool_obj.name,
+                    "description": tool_obj.description,
+                    "inputSchema": (
+                        tool_obj.inputSchema
+                        if isinstance(tool_obj.inputSchema, dict)
+                        else {}
+                    ),
+                    "required": tool_obj.inputSchema.get("required", []),
                 }
-            if not isinstance(tool, dict):
+            else:
+                tool_dict = tool_obj
+            if not isinstance(tool_dict, dict):
                 continue
-            tool_name = f"{mcp_name}_{tool.get('name', '')}"
-            desc = tool.get("description", "") or ""
+            tool_name = f"{mcp_name}_{tool_dict.get('name', '')}"
+            desc = tool_dict.get("description", "") or ""
             desc = desc.replace("\n", " ").replace("\r", " ").strip()
-            input_schema = tool.get("inputSchema", {})
+            input_schema = tool_dict.get("inputSchema", {})
             params = input_schema.get("properties", {}) if isinstance(input_schema, dict) else {}
-            required = tool.get("required", []) or []
+            required = tool_dict.get("required", []) or []
             param_str = (
                 "{ "
                 + ", ".join(