chore: 补全类型声明

astrbot/builtin_stars/astrbot/long_term_memory.py

@@ -23,7 +23,7 @@ def __init__(self, acm: AstrBotConfigManager, context: star.Context) -> None:
         self.session_chats = defaultdict(list)
         """记录群成员的群聊记录"""
 
-    def cfg(self, event: AstrMessageEvent):
+    def cfg(self, event: AstrMessageEvent) -> dict:
         cfg = self.context.get_config(umo=event.unified_msg_origin)
         try:
             max_cnt = int(cfg["provider_ltm_settings"]["group_message_max_cnt"])

astrbot/builtin_stars/astrbot/main.py

@@ -1,4 +1,5 @@
 import traceback
+from collections.abc import AsyncGenerator
 
 from astrbot.api import star
 from astrbot.api.event import AstrMessageEvent, filter
@@ -18,14 +19,16 @@ def __init__(self, context: star.Context) -> None:
         except BaseException as e:
             logger.error(f"聊天增强 err: {e}")
 
-    def ltm_enabled(self, event: AstrMessageEvent):
+    def ltm_enabled(self, event: AstrMessageEvent) -> bool:
         ltmse = self.context.get_config(umo=event.unified_msg_origin)[
             "provider_ltm_settings"
         ]
         return ltmse["group_icl_enable"] or ltmse["active_reply"]["enable"]
 
     @filter.platform_adapter_type(filter.PlatformAdapterType.ALL)
-    async def on_message(self, event: AstrMessageEvent):
+    async def on_message(
+        self, event: AstrMessageEvent
+    ) -> AsyncGenerator[ProviderRequest, None]:
         """群聊记忆增强"""
         has_image_or_plain = False
         for comp in event.message_obj.message:

astrbot/builtin_stars/builtin_commands/commands/conversation.py

@@ -20,7 +20,7 @@ class ConversationCommands:
     def __init__(self, context: star.Context) -> None:
         self.context = context
 
-    async def _get_current_persona_id(self, session_id):
+    async def _get_current_persona_id(self, session_id: str) -> str | None:
         curr = await self.context.conversation_manager.get_curr_conversation_id(
             session_id,
         )

astrbot/builtin_stars/builtin_commands/commands/help.py

@@ -11,7 +11,7 @@ class HelpCommand:
     def __init__(self, context: star.Context) -> None:
         self.context = context
 
-    async def _query_astrbot_notice(self):
+    async def _query_astrbot_notice(self) -> str:
         try:
             async with aiohttp.ClientSession(trust_env=True) as session:
                 async with session.get(

astrbot/builtin_stars/builtin_commands/commands/provider.py

@@ -5,6 +5,7 @@
 from astrbot.api import star
 from astrbot.api.event import AstrMessageEvent, MessageEventResult
 from astrbot.core.provider.entities import ProviderType
+from astrbot.core.provider.provider import Provider
 
 
 class ProviderCommands:
@@ -13,7 +14,7 @@ def __init__(self, context: star.Context) -> None:
 
     def _log_reachability_failure(
         self,
-        provider,
+        provider: Provider,
         provider_capability_type: ProviderType | None,
         err_code: str,
         err_reason: str,
@@ -28,7 +29,9 @@ def _log_reachability_failure(
             err_reason,
         )
 
-    async def _test_provider_capability(self, provider):
+    async def _test_provider_capability(
+        self, provider: Provider
+    ) -> tuple[bool, str | None, str | None]:
         """测试单个 provider 的可用性"""
         meta = provider.meta()
         provider_capability_type = meta.provider_type

astrbot/builtin_stars/session_controller/main.py

@@ -1,10 +1,13 @@
 import copy
+from collections.abc import AsyncGenerator
 from sys import maxsize
 
 import astrbot.api.message_components as Comp
 from astrbot.api import logger
 from astrbot.api.event import AstrMessageEvent, filter
 from astrbot.api.star import Context, Star
+from astrbot.core.message.message_event_result import MessageEventResult
+from astrbot.core.provider.entities import ProviderRequest
 from astrbot.core.utils.session_waiter import (
     FILTERS,
     USER_SESSIONS,
@@ -30,7 +33,9 @@ async def handle_session_control_agent(self, event: AstrMessageEvent) -> None:
                 event.stop_event()
 
     @filter.event_message_type(filter.EventMessageType.ALL, priority=maxsize - 1)
-    async def handle_empty_mention(self, event: AstrMessageEvent):
+    async def handle_empty_mention(
+        self, event: AstrMessageEvent
+    ) -> AsyncGenerator[MessageEventResult | ProviderRequest, None]:
         """实现了对只有一个 @ 的消息内容的处理"""
         try:
             messages = event.get_messages()

astrbot/builtin_stars/web_searcher/engines/bing.py

@@ -7,7 +7,7 @@ def __init__(self) -> None:
         self.base_urls = ["https://cn.bing.com", "https://www.bing.com"]
         self.headers.update({"User-Agent": USER_AGENT_BING})
 
-    def _set_selector(self, selector: str):
+    def _set_selector(self, selector: str) -> str:
         selectors = {
             "url": "div.b_attribution cite",
             "title": "h2",
@@ -17,7 +17,7 @@ def _set_selector(self, selector: str):
         }
         return selectors[selector]
 
-    async def _get_next_page(self, query) -> str:
+    async def _get_next_page(self, query: str) -> str:
         # if self.page == 1:
         #     await self._get_html(self.base_url)
         for base_url in self.base_urls:

astrbot/builtin_stars/web_searcher/engines/sogo.py

@@ -13,7 +13,7 @@ def __init__(self) -> None:
         self.base_url = "https://www.sogou.com"
         self.headers["User-Agent"] = random.choice(USER_AGENTS)
 
-    def _set_selector(self, selector: str):
+    def _set_selector(self, selector: str) -> str:
         selectors = {
             "url": "h3 > a",
             "title": "h3",
@@ -23,7 +23,7 @@ def _set_selector(self, selector: str):
         }
         return selectors[selector]
 
-    async def _get_next_page(self, query) -> str:
+    async def _get_next_page(self, query: str) -> str:
         url = f"{self.base_url}/web?query={query}"
         return await self._get_html(url, None)
 
@@ -38,7 +38,7 @@ async def search(self, query: str, num_results: int) -> list[SearchResult]:
                 result.url = await self._parse_url(result.url)
         return results
 
-    async def _parse_url(self, url) -> str:
+    async def _parse_url(self, url: str) -> str:
         html = await self._get_html(url)
         soup = BeautifulSoup(html, "html.parser")
         script = soup.find("script")

astrbot/builtin_stars/web_searcher/main.py

@@ -103,7 +103,7 @@ async def _process_search_result(
 
     async def _web_search_default(
         self,
-        query,
+        query: str,
         num_results: int = 5,
     ) -> list[SearchResult]:
         results = []

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:

astrbot/cli/commands/cmd_plug.py

@@ -5,6 +5,7 @@
 import click
 
 from ..utils import (
+    PluginInfo,
     PluginStatus,
     build_plug_list,
     check_astrbot_root,
@@ -28,7 +29,11 @@ def _get_data_path() -> Path:
     return (base / "data").resolve()
 
 
-def display_plugins(plugins, title=None, color=None) -> 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))
 
@@ -170,7 +175,7 @@ def remove(name: str) -> None:
     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"]

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,6 +20,16 @@ class PluginStatus(str, Enum):
     NOT_PUBLISHED = "未发布"
 
 
+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())
@@ -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,11 +26,10 @@ class HandoffTool(FunctionTool, Generic[TContext]):
     def __init__(
         self,
         agent: Agent[TContext],
-        parameters: dict | None = None,
+        parameters: ParametersType | None = None,
         tool_description: str | None = None,
-        **kwargs,
+        **kwargs: Unpack[HandoffInitKwargs],
     ) -> None:
-
         # Avoid passing duplicate `description` to the FunctionTool dataclass.
         # Some call sites (e.g. SubAgentOrchestrator) pass `description` via kwargs
         # to override what the main agent sees, while we also compute a default

astrbot/core/agent/mcp_client.py

@@ -4,6 +4,7 @@
 from datetime import timedelta
 from typing import Generic
 
+from mcp.types import CallToolResult
 from tenacity import (
     before_sleep_log,
     retry,
@@ -322,7 +323,7 @@ async def call_tool_with_reconnect(
             before_sleep=before_sleep_log(logger, logging.WARNING),
             reraise=True,
         )
-        async def _call_with_retry():
+        async def _call_with_retry() -> CallToolResult:
             if not self.session:
                 raise ValueError("MCP session is not available for MCP function tools.")
 
@@ -364,7 +365,11 @@ class MCPTool(FunctionTool, Generic[TContext]):
     """A function tool that calls an MCP service."""
 
     def __init__(
-        self, mcp_tool: mcp.Tool, mcp_client: MCPClient, mcp_server_name: str, **kwargs
+        self,
+        mcp_tool: mcp.Tool,
+        mcp_client: MCPClient,
+        mcp_server_name: str,
+        **kwargs: object,
     ) -> None:
         super().__init__(
             name=mcp_tool.name,
@@ -376,7 +381,9 @@ def __init__(
         self.mcp_server_name = mcp_server_name
 
     async def call(
-        self, context: ContextWrapper[TContext], **kwargs
+        self,
+        context: ContextWrapper[TContext],
+        **kwargs: object,
     ) -> mcp.types.CallToolResult:
         return await self.mcp_client.call_tool_with_reconnect(
             tool_name=self.mcp_tool.name,