chore: 补全类型声明

astrbot/cli/commands/cmd_conf.py

@@ -104,7 +104,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]:
@@ -118,7 +118,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:
@@ -127,7 +127,7 @@ def _get_nested_item(obj: dict[str, Any], path: str) -> Any:
 
 
 @click.group(name="conf")
-def conf():
+def conf() -> None:
     """配置管理命令
 
     支持的配置项:
@@ -149,7 +149,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:
         raise click.ClickException(f"不支持的配置项: {key}")
@@ -178,7 +178,7 @@ def set_config(key: str, value: str):
 
 @conf.command(name="get")
 @click.argument("key", required=False)
-def get_config(key: str | None = None):
+def get_config(key: str | None = None) -> None:
     """获取配置项的值，不提供key则显示所有可配置项"""
     config = _load_config()
 

astrbot/cli/commands/cmd_plug.py

@@ -5,6 +5,7 @@
 import click
 
 from ..utils import (
+    PluginInfo,
     PluginStatus,
     build_plug_list,
     check_astrbot_root,
@@ -15,7 +16,7 @@
 
 
 @click.group()
-def plug():
+def plug() -> None:
     """插件管理"""
 
 
@@ -28,7 +29,11 @@ def _get_data_path() -> Path:
     return (base / "data").resolve()
 
 
-def display_plugins(plugins, title=None, color=None):
+def display_plugins(
+    plugins: list[PluginInfo],
+    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))
 
@@ -45,7 +50,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
@@ -100,7 +105,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")
@@ -141,7 +146,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"
@@ -164,13 +169,13 @@ 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")
     plugin = next((p for p in plugins if p["name"] == name), None)
 
-    if not plugin or not plugin.get("local_path"):
+    if not plugin or not plugin["local_path"]:
         raise click.ClickException(f"插件 {name} 不存在或未安装")
 
     plugin_path = plugin["local_path"]
@@ -187,7 +192,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"
@@ -225,7 +230,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

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

astrbot/cli/utils/__init__.py

@@ -3,10 +3,17 @@
     check_dashboard,
     get_astrbot_root,
 )
-from .plugin import PluginStatus, build_plug_list, get_git_repo, manage_plugin
+from .plugin import (
+    PluginInfo,
+    PluginStatus,
+    build_plug_list,
+    get_git_repo,
+    manage_plugin,
+)
 from .version_comparator import VersionComparator
 
 __all__ = [
+    "PluginInfo",
     "PluginStatus",
     "VersionComparator",
     "build_plug_list",

astrbot/cli/utils/plugin.py

@@ -3,6 +3,7 @@
 from enum import Enum
 from io import BytesIO
 from pathlib import Path
+from typing import TypedDict
 from zipfile import ZipFile
 
 import click
@@ -19,7 +20,17 @@ class PluginStatus(str, Enum):
     NOT_PUBLISHED = "未发布"
 
 
-def get_git_repo(url: str, target_path: Path, proxy: str | None = None):
+class PluginInfo(TypedDict):
+    name: str
+    desc: str
+    version: str
+    author: str
+    repo: str
+    status: PluginStatus
+    local_path: str | None
+
+
+def get_git_repo(url: str, target_path: Path, proxy: str | None = None) -> None:
     """从 Git 仓库下载代码并解压到指定路径"""
     temp_dir = Path(tempfile.mkdtemp())
     try:
@@ -102,18 +113,18 @@ def load_yaml_metadata(plugin_dir: Path) -> dict:
     return {}
 
 
-def build_plug_list(plugins_dir: Path) -> list:
+def build_plug_list(plugins_dir: Path) -> list[PluginInfo]:
     """构建插件列表，包含本地和在线插件信息
 
     Args:
         plugins_dir (Path): 插件目录路径
 
     Returns:
-        list: 包含插件信息的字典列表
+        list[PluginInfo]: 包含插件信息的字典列表
 
     """
     # 获取本地插件信息
-    result = []
+    result: list[PluginInfo] = []
     if plugins_dir.exists():
         for plugin_name in [d.name for d in plugins_dir.glob("*") if d.is_dir()]:
             plugin_dir = plugins_dir / plugin_name
@@ -141,7 +152,7 @@ def build_plug_list(plugins_dir: Path) -> list:
                 )
 
     # 获取在线插件列表
-    online_plugins = []
+    online_plugins: list[PluginInfo] = []
     try:
         with httpx.Client() as client:
             resp = client.get("https://api.soulter.top/astrbot/plugins")
@@ -191,7 +202,7 @@ def build_plug_list(plugins_dir: Path) -> list:
 
 
 def manage_plugin(
-    plugin: dict,
+    plugin: PluginInfo,
     plugins_dir: Path,
     is_update: bool = False,
     proxy: str | None = None,
@@ -209,7 +220,7 @@ def manage_plugin(
     repo_url = plugin["repo"]
 
     # 如果是更新且有本地路径，直接使用本地路径
-    if is_update and plugin.get("local_path"):
+    if is_update and plugin["local_path"]:
         target_path = Path(plugin["local_path"])
     else:
         target_path = plugins_dir / plugin_name

astrbot/cli/utils/version_comparator.py

@@ -15,7 +15,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,
@@ -77,7 +77,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,8 +1,23 @@
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator, Awaitable, Callable
 from typing import Generic
 
+from typing_extensions import TypedDict, Unpack
+
+from astrbot.core.message.message_event_result import MessageEventResult
+
 from .agent import Agent
 from .run_context import TContext
-from .tool import FunctionTool
+from .tool import FunctionTool, ParametersType
+
+
+class HandoffInitKwargs(TypedDict, total=False):
+    handler: (
+        Callable[..., Awaitable[str | None] | AsyncGenerator[MessageEventResult]] | None
+    )
+    handler_module_path: str | None
+    active: bool
 
 
 class HandoffTool(FunctionTool, Generic[TContext]):
@@ -11,9 +26,9 @@ class HandoffTool(FunctionTool, Generic[TContext]):
     def __init__(
         self,
         agent: Agent[TContext],
-        parameters: dict | None = None,
-        **kwargs,
-    ):
+        parameters: ParametersType | None = None,
+        **kwargs: Unpack[HandoffInitKwargs],
+    ) -> None:
         self.agent = agent
         super().__init__(
             name=f"transfer_to_{agent.name}",

astrbot/core/agent/hooks.py

@@ -9,22 +9,22 @@
 
 
 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: ...