可选后端，实现前后端分离

.python-version

@@ -1 +1 @@
-3.10
+3.10

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/cli/commands/cmd_run.py

@@ -27,9 +27,17 @@ async def run_astrbot(astrbot_root: Path):
 
 
 @click.option("--reload", "-r", is_flag=True, help="插件自动重载")
-@click.option("--port", "-p", help="Astrbot Dashboard端口", required=False, type=str)
+@click.option(
+    "--host", "-H", help="Astrbot Dashboard Host,默认::", required=False, type=str
+)
+@click.option(
+    "--port", "-p", help="Astrbot Dashboard端口,默认6185", required=False, type=str
+)
+@click.option(
+    "--backend-only", is_flag=True, default=False, help="禁用WEBUI,仅启动后端"
+)
 @click.command()
-def run(reload: bool, port: str) -> None:
+def run(reload: bool, host: str, port: str, backend_only: bool) -> None:
     """运行 AstrBot"""
     try:
         os.environ["ASTRBOT_CLI"] = "1"
@@ -43,8 +51,9 @@ def run(reload: bool, port: str) -> None:
         os.environ["ASTRBOT_ROOT"] = str(astrbot_root)
         sys.path.insert(0, str(astrbot_root))
 
-        if port:
-            os.environ["DASHBOARD_PORT"] = port
+        os.environ["DASHBOARD_PORT"] = port or "6185"
+        os.environ["DASHBOARD_HOST"] = host or "::"
+        os.environ["DASHBOARD_ENABLE"] = str(not backend_only)
 
         if reload:
             click.echo("启用插件自动重载")

astrbot/core/agent/tool.py

@@ -246,8 +246,18 @@ def convert_schema(schema: dict) -> dict:
 
             result = {}
 
-            if "type" in schema and schema["type"] in supported_types:
-                result["type"] = schema["type"]
+            # Avoid side effects by not modifying the original schema
+            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/config/default.py

@@ -182,7 +182,7 @@
         "username": "astrbot",
         "password": "77b90590a8945a7d36c963981a307dc9",
         "jwt_secret": "",
-        "host": "0.0.0.0",
+        "host": "::",
         "port": 6185,
         "disable_access_log": True,
     },
@@ -273,14 +273,14 @@ class ChatProviderTemplate(TypedDict):
                         "is_sandbox": False,
                         "unified_webhook_mode": True,
                         "webhook_uuid": "",
-                        "callback_server_host": "0.0.0.0",
+                        "callback_server_host": "::",
                         "port": 6196,
                     },
                     "OneBot v11": {
                         "id": "default",
                         "type": "aiocqhttp",
                         "enable": False,
-                        "ws_reverse_host": "0.0.0.0",
+                        "ws_reverse_host": "::",
                         "ws_reverse_port": 6199,
                         "ws_reverse_token": "",
                     },
@@ -295,7 +295,7 @@ class ChatProviderTemplate(TypedDict):
                         "api_base_url": "https://api.weixin.qq.com/cgi-bin/",
                         "unified_webhook_mode": True,
                         "webhook_uuid": "",
-                        "callback_server_host": "0.0.0.0",
+                        "callback_server_host": "::",
                         "port": 6194,
                         "active_send_mode": False,
                     },
@@ -311,7 +311,7 @@ class ChatProviderTemplate(TypedDict):
                         "api_base_url": "https://qyapi.weixin.qq.com/cgi-bin/",
                         "unified_webhook_mode": True,
                         "webhook_uuid": "",
-                        "callback_server_host": "0.0.0.0",
+                        "callback_server_host": "::",
                         "port": 6195,
                     },
                     "企业微信智能机器人": {
@@ -325,7 +325,7 @@ class ChatProviderTemplate(TypedDict):
                         "encoding_aes_key": "",
                         "unified_webhook_mode": True,
                         "webhook_uuid": "",
-                        "callback_server_host": "0.0.0.0",
+                        "callback_server_host": "::",
                         "port": 6198,
                     },
                     "飞书(Lark)": {
@@ -399,7 +399,7 @@ class ChatProviderTemplate(TypedDict):
                         "slack_connection_mode": "socket",  # webhook, socket
                         "unified_webhook_mode": True,
                         "webhook_uuid": "",
-                        "slack_webhook_host": "0.0.0.0",
+                        "slack_webhook_host": "::",
                         "slack_webhook_port": 6197,
                         "slack_webhook_path": "/astrbot-slack-webhook/callback",
                     },

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 @@ def check(self, content: str) -> tuple[bool, str]:
         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/aiocqhttp/aiocqhttp_platform_adapter.py

@@ -418,9 +418,9 @@ async def _convert_handle_message_event(
     def run(self) -> Awaitable[Any]:
         if not self.host or not self.port:
             logger.warning(
-                "aiocqhttp: 未配置 ws_reverse_host 或 ws_reverse_port，将使用默认值：http://0.0.0.0:6199",
+                "aiocqhttp: 未配置 ws_reverse_host 或 ws_reverse_port，将使用默认值：http://[::]:6199",
             )
-            self.host = "0.0.0.0"
+            self.host = "::"
             self.port = 6199
 
         coro = self.bot.run_task(

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

@@ -444,9 +444,20 @@ async def dynamic_callback(
                 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)
-            abm.group_id = self._get_channel_id(ctx.channel)
+            if channel is not None:
+                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/platform/sources/qqofficial_webhook/qo_webhook_server.py

@@ -19,7 +19,7 @@ def __init__(self, config: dict, event_queue: asyncio.Queue, botpy_client: Clien
         self.secret = config["secret"]
         self.port = config.get("port", 6196)
         self.is_sandbox = config.get("is_sandbox", False)
-        self.callback_server_host = config.get("callback_server_host", "0.0.0.0")
+        self.callback_server_host = config.get("callback_server_host", "::")
 
         if isinstance(self.port, str):
             self.port = int(self.port)

astrbot/core/platform/sources/slack/client.py

@@ -23,7 +23,7 @@ def __init__(
         self,
         web_client: AsyncWebClient,
         signing_secret: str,
-        host: str = "0.0.0.0",
+        host: str = "::",
         port: int = 3000,
         path: str = "/slack/events",
         event_handler: Callable | None = None,

astrbot/core/platform/sources/slack/slack_adapter.py

@@ -47,7 +47,7 @@ def __init__(
         self.signing_secret = platform_config.get("signing_secret")
         self.connection_mode = platform_config.get("slack_connection_mode", "socket")
         self.unified_webhook_mode = platform_config.get("unified_webhook_mode", False)
-        self.webhook_host = platform_config.get("slack_webhook_host", "0.0.0.0")
+        self.webhook_host = platform_config.get("slack_webhook_host", "::")
         self.webhook_port = platform_config.get("slack_webhook_port", 3000)
         self.webhook_path = platform_config.get(
             "slack_webhook_path",

astrbot/core/platform/sources/wecom/wecom_adapter.py

@@ -42,7 +42,7 @@ class WecomServer:
     def __init__(self, event_queue: asyncio.Queue, config: dict):
         self.server = quart.Quart(__name__)
         self.port = int(cast(str, config.get("port")))
-        self.callback_server_host = config.get("callback_server_host", "0.0.0.0")
+        self.callback_server_host = config.get("callback_server_host", "::")
         self.server.add_url_rule(
             "/callback/command",
             view_func=self.verify,

astrbot/core/platform/sources/wecom_ai_bot/wecomai_adapter.py

@@ -111,7 +111,7 @@ def __init__(
         self.token = self.config["token"]
         self.encoding_aes_key = self.config["encoding_aes_key"]
         self.port = int(self.config["port"])
-        self.host = self.config.get("callback_server_host", "0.0.0.0")
+        self.host = self.config.get("callback_server_host", "::")
         self.bot_name = self.config.get("wecom_ai_bot_name", "")
         self.initial_respond_text = self.config.get(
             "wecomaibot_init_respond_text",

astrbot/core/platform/sources/weixin_official_account/weixin_offacc_adapter.py

@@ -38,7 +38,7 @@ class WeixinOfficialAccountServer:
     def __init__(self, event_queue: asyncio.Queue, config: dict):
         self.server = quart.Quart(__name__)
         self.port = int(cast(int | str, config.get("port")))
-        self.callback_server_host = config.get("callback_server_host", "0.0.0.0")
+        self.callback_server_host = config.get("callback_server_host", "::")
         self.token = config.get("token")
         self.encoding_aes_key = config.get("encoding_aes_key")
         self.appid = config.get("appid")

astrbot/core/provider/sources/fishaudio_tts_api_source.py

@@ -63,7 +63,7 @@ def __init__(
         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/core/utils/io.py

@@ -1,3 +1,4 @@
+import asyncio
 import base64
 import logging
 import os
@@ -7,6 +8,7 @@
 import time
 import uuid
 import zipfile
+from ipaddress import IPv4Address, IPv6Address, ip_address
 from pathlib import Path
 
 import aiohttp
@@ -217,18 +219,51 @@ def file_to_base64(file_path: str) -> str:
     return "base64://" + base64_str
 
 
-def get_local_ip_addresses():
+def get_local_ip_addresses() -> list[IPv4Address | IPv6Address]:
     net_interfaces = psutil.net_if_addrs()
-    network_ips = []
+    network_ips: list[IPv4Address | IPv6Address] = []
 
-    for interface, addrs in net_interfaces.items():
+    for _, addrs in net_interfaces.items():
         for addr in addrs:
-            if addr.family == socket.AF_INET:  # 使用 socket.AF_INET 代替 psutil.AF_INET
-                network_ips.append(addr.address)
+            if addr.family == socket.AF_INET:
+                network_ips.append(ip_address(addr.address))
+            elif addr.family == socket.AF_INET6:
+                # 过滤掉 IPv6 的 link-local 地址（fe80:...）
+                # 用这个不如用::1
+                ip = ip_address(addr.address.split("%")[0])  # 处理带 zone index 的情况
+                network_ips.append(ip)
 
     return network_ips
 
 
+async def get_public_ip_address() -> list[IPv4Address | IPv6Address]:
+    urls = [
+        "https://api64.ipify.org",
+        "https://ident.me",
+        "https://ifconfig.me",
+        "https://icanhazip.com",
+    ]
+    found_ips: dict[int, IPv4Address | IPv6Address] = {}
+
+    async def fetch(session: aiohttp.ClientSession, url: str):
+        try:
+            async with session.get(url, timeout=3) as resp:
+                if resp.status == 200:
+                    raw_ip = (await resp.text()).strip()
+                    ip = ip_address(raw_ip)
+                    if ip.version not in found_ips:
+                        found_ips[ip.version] = ip
+        except Exception:
+            pass
+
+    async with aiohttp.ClientSession() as session:
+        tasks = [fetch(session, url) for url in urls]
+        await asyncio.gather(*tasks)
+
+    # 返回找到的所有 IP 对象列表
+    return list(found_ips.values())
+
+
 async def get_dashboard_version():
     dist_dir = os.path.join(get_astrbot_data_path(), "dist")
     if os.path.exists(dist_dir):

astrbot/dashboard/routes/cron.py

@@ -23,7 +23,7 @@ def __init__(
         ]
         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 os
 import traceback
 import uuid
+from typing import Any
 
 import aiofiles
 from quart import request
@@ -75,7 +76,7 @@ def _init_task(self, task_id: str, status: str = "pending") -> None:
         }
 
     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/routes/route.py

@@ -1,6 +1,6 @@
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
 
-from quart import Quart
+from quart import Quart, jsonify
 
 from astrbot.core.config.astrbot_config import AstrBotConfig
 
@@ -57,3 +57,6 @@ def ok(self, data: dict | list | None = None, message: str | None = None):
         self.data = data
         self.message = message
         return self
+
+    def to_json(self):
+        return jsonify(asdict(self))