chore: 补全类型声明

astrbot/cli/commands/cmd_conf.py

@@ -2,7 +2,9 @@
 import click
 import hashlib
 import zoneinfo
-from typing import Any, Callable
+from typing import Any
+
+from collections.abc import Callable
 from ..utils import get_astrbot_root, check_astrbot_root
 
 
@@ -100,7 +102,7 @@ def _save_config(config: dict[str, Any]) -> None:
     )
 
 
-def _set_nested_item(obj: dict[str, Any], path: str, value: Any) -> None:
+def _set_nested_item(obj: dict[str, Any], path: str, value: object) -> None:
     """设置嵌套字典中的值"""
     parts = path.split(".")
     for part in parts[:-1]:
@@ -114,7 +116,7 @@ def _set_nested_item(obj: dict[str, Any], path: str, value: Any) -> None:
     obj[parts[-1]] = value
 
 
-def _get_nested_item(obj: dict[str, Any], path: str) -> Any:
+def _get_nested_item(obj: dict[str, Any], path: str) -> object:
     """获取嵌套字典中的值"""
     parts = path.split(".")
     for part in parts:
@@ -123,7 +125,7 @@ def _get_nested_item(obj: dict[str, Any], path: str) -> Any:
 
 
 @click.group(name="conf")
-def conf():
+def conf() -> None:
     """配置管理命令
 
     支持的配置项:
@@ -146,7 +148,7 @@ def conf():
 @conf.command(name="set")
 @click.argument("key")
 @click.argument("value")
-def set_config(key: str, value: str):
+def set_config(key: str, value: str) -> None:
     """设置配置项的值"""
     if key not in CONFIG_VALIDATORS.keys():
         raise click.ClickException(f"不支持的配置项: {key}")
@@ -175,7 +177,7 @@ def set_config(key: str, value: str):
 
 @conf.command(name="get")
 @click.argument("key", required=False)
-def get_config(key: str = None):
+def get_config(key: str | None = None) -> None:
     """获取配置项的值，不提供key则显示所有可配置项"""
     config = _load_config()
 

astrbot/cli/commands/cmd_init.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 import asyncio
 
 import click
@@ -6,7 +8,7 @@
 from ..utils import check_dashboard, get_astrbot_root
 
 
-async def initialize_astrbot(astrbot_root) -> None:
+async def initialize_astrbot(astrbot_root: Path) -> None:
     """执行 AstrBot 初始化逻辑"""
     dot_astrbot = astrbot_root / ".astrbot"
 

astrbot/cli/commands/cmd_plug.py

@@ -16,7 +16,7 @@
 
 
 @click.group()
-def plug():
+def plug() -> None:
     """插件管理"""
     pass
 
@@ -30,7 +30,11 @@ def _get_data_path() -> Path:
     return (base / "data").resolve()
 
 
-def display_plugins(plugins, title=None, color=None):
+def display_plugins(
+    plugins: list[dict],
+    title: str | None = None,
+    color: int | tuple[int, int, int] | str | None = None,
+) -> None:
     if title:
         click.echo(click.style(title, fg=color, bold=True))
 
@@ -47,7 +51,7 @@ def display_plugins(plugins, title=None, color=None):
 
 @plug.command()
 @click.argument("name")
-def new(name: str):
+def new(name: str) -> None:
     """创建新插件"""
     base_path = _get_data_path()
     plug_path = base_path / "plugins" / name
@@ -86,7 +90,7 @@ def new(name: str):
         f.write(f"# {name}\n\n{desc}\n\n# 支持\n\n[帮助文档](https://astrbot.app)\n")
 
     # 重写 main.py
-    with open(plug_path / "main.py", "r", encoding="utf-8") as f:
+    with open(plug_path / "main.py", encoding="utf-8") as f:
         content = f.read()
 
     new_content = content.replace(
@@ -102,7 +106,7 @@ def new(name: str):
 
 @plug.command()
 @click.option("--all", "-a", is_flag=True, help="列出未安装的插件")
-def list(all: bool):
+def list(all: bool) -> None:
     """列出插件"""
     base_path = _get_data_path()
     plugins = build_plug_list(base_path / "plugins")
@@ -143,7 +147,7 @@ def list(all: bool):
 @plug.command()
 @click.argument("name")
 @click.option("--proxy", help="代理服务器地址")
-def install(name: str, proxy: str | None):
+def install(name: str, proxy: str | None) -> None:
     """安装插件"""
     base_path = _get_data_path()
     plug_path = base_path / "plugins"
@@ -166,7 +170,7 @@ def install(name: str, proxy: str | None):
 
 @plug.command()
 @click.argument("name")
-def remove(name: str):
+def remove(name: str) -> None:
     """卸载插件"""
     base_path = _get_data_path()
     plugins = build_plug_list(base_path / "plugins")
@@ -189,7 +193,7 @@ def remove(name: str):
 @plug.command()
 @click.argument("name", required=False)
 @click.option("--proxy", help="Github代理地址")
-def update(name: str, proxy: str | None):
+def update(name: str, proxy: str | None) -> None:
     """更新插件"""
     base_path = _get_data_path()
     plug_path = base_path / "plugins"
@@ -227,7 +231,7 @@ def update(name: str, proxy: str | None):
 
 @plug.command()
 @click.argument("query")
-def search(query: str):
+def search(query: str) -> None:
     """搜索插件"""
     base_path = _get_data_path()
     plugins = build_plug_list(base_path / "plugins")

astrbot/cli/commands/cmd_run.py

@@ -11,7 +11,7 @@
 from ..utils import check_dashboard, check_astrbot_root, get_astrbot_root
 
 
-async def run_astrbot(astrbot_root: Path):
+async def run_astrbot(astrbot_root: Path) -> None:
     """运行 AstrBot"""
     from astrbot.core import logger, LogManager, LogBroker, db_helper
     from astrbot.core.initial_loader import InitialLoader

astrbot/cli/utils/plugin.py

@@ -19,7 +19,7 @@ class PluginStatus(str, Enum):
     NOT_PUBLISHED = "未发布"
 
 
-def get_git_repo(url: str, target_path: Path, proxy: str | None = None):
+def get_git_repo(url: str, target_path: Path, proxy: str | None = None) -> None:
     """从 Git 仓库下载代码并解压到指定路径"""
     temp_dir = Path(tempfile.mkdtemp())
     try:

astrbot/cli/utils/version_comparator.py

@@ -17,7 +17,7 @@ def compare_version(v1: str, v2: str) -> int:
         v1 = v1.lower().replace("v", "")
         v2 = v2.lower().replace("v", "")
 
-        def split_version(version):
+        def split_version(version: str) -> tuple[list[int], list[int | str] | None]:
             match = re.match(
                 r"^([0-9]+(?:\.[0-9]+)*)(?:-([0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+(.+))?$",
                 version,
@@ -79,7 +79,7 @@ def split_version(version):
         return 0  # 数字部分和预发布标签都相同
 
     @staticmethod
-    def _split_prerelease(prerelease):
+    def _split_prerelease(prerelease: str) -> list[int | str] | None:
         if not prerelease:
             return None
         parts = prerelease.split(".")

astrbot/core/agent/handoff.py

@@ -1,4 +1,4 @@
-from typing import Generic
+from typing import Any, Generic
 from .tool import FunctionTool
 from .agent import Agent
 from .run_context import TContext
@@ -8,8 +8,11 @@ class HandoffTool(FunctionTool, Generic[TContext]):
     """Handoff tool for delegating tasks to another agent."""
 
     def __init__(
-        self, agent: Agent[TContext], parameters: dict | None = None, **kwargs
-    ):
+        self,
+        agent: Agent[TContext],
+        parameters: dict | None = None,
+        **kwargs: Any,  # noqa: ANN401
+    ) -> None:
         self.agent = agent
         super().__init__(
             name=f"transfer_to_{agent.name}",

astrbot/core/agent/hooks.py

@@ -8,20 +8,20 @@
 
 @dataclass
 class BaseAgentRunHooks(Generic[TContext]):
-    async def on_agent_begin(self, run_context: ContextWrapper[TContext]): ...
+    async def on_agent_begin(self, run_context: ContextWrapper[TContext]) -> None: ...
     async def on_tool_start(
         self,
         run_context: ContextWrapper[TContext],
         tool: FunctionTool,
         tool_args: dict | None,
-    ): ...
+    ) -> None: ...
     async def on_tool_end(
         self,
         run_context: ContextWrapper[TContext],
         tool: FunctionTool,
         tool_args: dict | None,
         tool_result: mcp.types.CallToolResult | None,
-    ): ...
+    ) -> None: ...
     async def on_agent_done(
         self, run_context: ContextWrapper[TContext], llm_response: LLMResponse
-    ): ...
+    ) -> None: ...

astrbot/core/agent/mcp_client.py

@@ -1,7 +1,6 @@
 import asyncio
 import logging
 from datetime import timedelta
-from typing import Optional
 from contextlib import AsyncExitStack
 from astrbot import logger
 from astrbot.core.utils.log_pipe import LogPipe
@@ -94,9 +93,9 @@ async def _quick_test_mcp_connection(config: dict) -> tuple[bool, str]:
 
 
 class MCPClient:
-    def __init__(self):
+    def __init__(self) -> None:
         # Initialize session and client objects
-        self.session: Optional[mcp.ClientSession] = None
+        self.session: mcp.ClientSession | None = None
         self.exit_stack = AsyncExitStack()
 
         self.name: str | None = None
@@ -105,7 +104,7 @@ def __init__(self):
         self.server_errlogs: list[str] = []
         self.running_event = asyncio.Event()
 
-    async def connect_to_server(self, mcp_server_config: dict, name: str):
+    async def connect_to_server(self, mcp_server_config: dict, name: str) -> None:
         """连接到 MCP 服务器
 
         如果 `url` 参数存在：
@@ -118,7 +117,7 @@ async def connect_to_server(self, mcp_server_config: dict, name: str):
         """
         cfg = _prepare_config(mcp_server_config.copy())
 
-        def logging_callback(msg: str):
+        def logging_callback(msg: str) -> None:
             # 处理 MCP 服务的错误日志
             print(f"MCP Server {name} Error: {msg}")
             self.server_errlogs.append(msg)
@@ -188,7 +187,7 @@ def logging_callback(msg: str):
                 **cfg,
             )
 
-            def callback(msg: str):
+            def callback(msg: str) -> None:
                 # 处理 MCP 服务的错误日志
                 self.server_errlogs.append(msg)
 
@@ -218,7 +217,7 @@ async def list_tools_and_save(self) -> mcp.ListToolsResult:
         self.tools = response.tools
         return response
 
-    async def cleanup(self):
+    async def cleanup(self) -> None:
         """Clean up resources"""
         await self.exit_stack.aclose()
         self.running_event.set()  # Set the running event to indicate cleanup is done

astrbot/core/agent/runners/base.py

@@ -26,7 +26,7 @@ async def reset(
         run_context: ContextWrapper[TContext],
         tool_executor: BaseFunctionToolExecutor[TContext],
         agent_hooks: BaseAgentRunHooks[TContext],
-        **kwargs: T.Any,
+        **kwargs: T.Any,  # noqa: ANN401
     ) -> None:
         """
         Reset the agent to its initial state.

astrbot/core/agent/runners/tool_loop_agent_runner.py

@@ -69,7 +69,7 @@ async def _iter_llm_responses(self) -> T.AsyncGenerator[LLMResponse, None]:
             yield await self.provider.text_chat(**self.req.__dict__)
 
     @override
-    async def step(self):
+    async def step(self) -> T.AsyncGenerator[AgentResponse, None]:
         """
         Process a single step of the agent.
         This method should return the result of the step.