refactor: extract helper method to reduce code duplication

Co-authored-by: Soulter <37870767+Soulter@users.noreply.github.com>

README.md

@@ -264,8 +264,9 @@ pre-commit install
 
 <div align="center">
 
+_陪伴与能力从来不应该是对立面。我们希望创造的是一个既能理解情绪、给予陪伴，也能可靠完成工作的机器人。_
+
 _私は、高性能ですから!_
 
-陪伴与能力从来不应该是对立面。我们希望创造的是一个既能理解情绪、给予陪伴，也能可靠完成工作的机器人。
+<img src="https://files.astrbot.app/watashiwa-koseino-desukara.gif" width="100"/>
-
 

astrbot/builtin_stars/web_searcher/main.py

@@ -49,6 +49,24 @@ class Main(star.Star):
         self.sogo_search = Sogo()
         self.baidu_initialized = False
 
+        # Deactivate built-in web search tools if web_search is disabled
+        # This allows MCP to provide custom web_search tools
+        websearch_enable = (provider_settings or {}).get("web_search", False)
+        if not websearch_enable:
+            self._set_tools_active(False)
+
+    def _set_tools_active(self, active: bool) -> None:
+        """Set the active status of all built-in web search tools.
+
+        Args:
+            active: True to activate tools, False to deactivate them
+        """
+        func_tool_mgr = self.context.get_llm_tool_manager()
+        for tool_name in self.TOOLS:
+            tool = func_tool_mgr.get_func(tool_name)
+            if tool:
+                tool.active = active
+
     async def _tidy_text(self, text: str) -> str:
         """清理文本，去除空格、换行符等"""
         return text.strip().replace("\n", " ").replace("\r", " ").replace("  ", " ")
@@ -394,6 +412,10 @@ class Main(star.Star):
         websearch_enable = prov_settings.get("web_search", False)
         provider = prov_settings.get("websearch_provider", "default")
 
+        # Globally activate/deactivate built-in web search tools based on config
+        # This allows MCP to provide custom web_search tools when built-in is disabled
+        self._set_tools_active(websearch_enable)
+
         tool_set = req.func_tool
         if isinstance(tool_set, FunctionToolManager):
             req.func_tool = tool_set.get_full_tool_set()

astrbot/core/agent/tool.py

@@ -246,8 +246,18 @@ class ToolSet:
 
             result = {}
 
-            if "type" in schema and schema["type"] in supported_types:
+            # Avoid side effects by not modifying the original schema
-                result["type"] = schema["type"]
+            origin_type = schema.get("type")
+            target_type = origin_type
+
+            # Compatibility fix: Gemini API expects 'type' to be a string (enum),
+            # but standard JSON Schema (MCP) allows lists (e.g. ["string", "null"]).
+            # We fallback to the first non-null type.
+            if isinstance(origin_type, list):
+                target_type = next((t for t in origin_type if t != "null"), "string")
+
+            if target_type in supported_types:
+                result["type"] = target_type
                 if "format" in schema and schema["format"] in supported_formats.get(
                     result["type"],
                     set(),

astrbot/core/pipeline/content_safety_check/strategies/baidu_aip.py

@@ -1,5 +1,7 @@
 """使用此功能应该先 pip install baidu-aip"""
 
+from typing import Any, cast
+
 from aip import AipContentCensor
 
 from . import ContentSafetyStrategy
@@ -23,7 +25,8 @@ class BaiduAipStrategy(ContentSafetyStrategy):
         count = len(res["data"])
         parts = [f"百度审核服务发现 {count} 处违规：\n"]
         for i in res["data"]:
-            parts.append(f"{i['msg']}；\n")
+            # 百度 AIP 返回结构是动态 dict；类型检查时 i 可能被推断为序列，转成 dict 后用 get 取字段
+            parts.append(f"{cast(dict[str, Any], i).get('msg', '')}；\n")
         parts.append("\n判断结果：" + res["conclusion"])
         info = "".join(parts)
         return False, info

astrbot/core/platform/message_session.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from astrbot.core.platform.message_type import MessageType
 
@@ -13,7 +13,7 @@ class MessageSession:
     """平台适配器实例的唯一标识符。自 AstrBot v4.0.0 起，该字段实际为 platform_id。"""
     message_type: MessageType
     session_id: str
-    platform_id: str | None = None
+    platform_id: str = field(init=False)
 
     def __str__(self):
         return f"{self.platform_id}:{self.message_type.value}:{self.session_id}"

astrbot/core/platform/sources/discord/discord_platform_adapter.py

@@ -444,9 +444,20 @@ class DiscordPlatformAdapter(Platform):
                 logger.warning(f"[Discord] 指令 '{cmd_name}' defer 失败: {e}")
 
             # 2. 构建 AstrBotMessage
+            channel = ctx.channel
             abm = AstrBotMessage()
-            abm.type = self._get_message_type(ctx.channel, ctx.guild_id)
+            if channel is not None:
-            abm.group_id = self._get_channel_id(ctx.channel)
+                abm.type = self._get_message_type(channel, ctx.guild_id)
+                abm.group_id = self._get_channel_id(channel)
+            else:
+                # 防守式兜底：channel 取不到时，仍能根据 guild_id/channel_id 推断会话信息
+                abm.type = (
+                    MessageType.GROUP_MESSAGE
+                    if ctx.guild_id is not None
+                    else MessageType.FRIEND_MESSAGE
+                )
+                abm.group_id = str(ctx.channel_id)
+
             abm.message_str = message_str_for_filter
             abm.sender = MessageMember(
                 user_id=str(ctx.author.id),

astrbot/core/provider/sources/fishaudio_tts_api_source.py

@@ -63,7 +63,7 @@ class ProviderFishAudioTTSAPI(TTSProvider):
         self.headers = {
             "Authorization": f"Bearer {self.chosen_api_key}",
         }
-        self.set_model(provider_config.get("model", None))
+        self.set_model(provider_config.get("model", ""))
 
     async def _get_reference_id_by_character(self, character: str) -> str | None:
         """获取角色的reference_id

astrbot/dashboard/routes/cron.py

@@ -23,7 +23,7 @@ class CronRoute(Route):
         ]
         self.register_routes()
 
-    def _serialize_job(self, job):
+    def _serialize_job(self, job) -> dict:
         data = job.model_dump() if hasattr(job, "model_dump") else job.__dict__
         for k in ["created_at", "updated_at", "last_run_at", "next_run_time"]:
             if isinstance(data.get(k), datetime):

astrbot/dashboard/routes/knowledge_base.py

@@ -4,6 +4,7 @@ import asyncio
 import os
 import traceback
 import uuid
+from typing import Any
 
 import aiofiles
 from quart import request
@@ -75,7 +76,7 @@ class KnowledgeBaseRoute(Route):
         }
 
     def _set_task_result(
-        self, task_id: str, status: str, result: any = None, error: str | None = None
+        self, task_id: str, status: str, result: Any = None, error: str | None = None
     ) -> None:
         self.upload_tasks[task_id] = {
             "status": status,

astrbot/dashboard/server.py

@@ -2,14 +2,13 @@ import asyncio
 import logging
 import os
 import socket
-from typing import cast
+from typing import Protocol, cast
 
 import jwt
 import psutil
 from flask.json.provider import DefaultJSONProvider
 from hypercorn.asyncio import serve
 from hypercorn.config import Config as HyperConfig
-from psutil._common import addr as psutil_addr
 from quart import Quart, g, jsonify, request
 from quart.logging import default_handler
 
@@ -29,6 +28,11 @@ from .routes.session_management import SessionManagementRoute
 from .routes.subagent import SubAgentRoute
 from .routes.t2i import T2iRoute
 
+
+class _AddrWithPort(Protocol):
+    port: int
+
+
 APP: Quart
 
 
@@ -168,7 +172,7 @@ class AstrBotDashboard:
         """获取占用端口的进程详细信息"""
         try:
             for conn in psutil.net_connections(kind="inet"):
-                if cast(psutil_addr, conn.laddr).port == port:
+                if cast(_AddrWithPort, conn.laddr).port == port:
                     try:
                         process = psutil.Process(conn.pid)
                         # 获取详细信息

scripts/astrbot.service

@@ -0,0 +1,15 @@
+[Unit]
+Description=AstrBot Service
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+WorkingDirectory=%h/.local/share/astrbot
+ExecStart=/usr/bin/sh -c '/usr/bin/astrbot run || { /usr/bin/astrbot init && /usr/bin/astrbot run; }'
+Restart=on-failure
+RestartSec=5
+Environment=PYTHONUNBUFFERED=1
+
+[Install]
+WantedBy=default.target