Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox_Bot into dev

README.md

@@ -85,6 +85,7 @@
 - 🧠 **拓展记忆系统** - 支持瞬时记忆等多种记忆
 - 🎪 **完善的 Event** - 支持动态事件注册和处理器订阅，并实现了聚合结果管理
 - 🔍 **内嵌魔改插件** - 内置联网搜索等诸多功能，等你来探索
+- 🔌 **MCP 协议支持** - 集成 Model Context Protocol，支持外部工具服务器连接（仅 Streamable HTTP）
 - 🌟 **还有更多** - 请参阅详细修改 [commits](https://github.com/MoFox-Studio/MoFox_Bot/commits)
  
 </td>

plugins/hello_world_plugin/plugin.py

@@ -19,8 +19,8 @@ from src.plugin_system import (
     ToolParamType,
     register_plugin,
 )
-from src.plugin_system.base.component_types import InjectionRule,InjectionType
 from src.plugin_system.base.base_event import HandlerResult
+from src.plugin_system.base.component_types import InjectionRule, InjectionType
 
 logger = get_logger("hello_world_plugin")
 

pyproject.toml

@@ -74,6 +74,7 @@ dependencies = [
     "aiosqlite>=0.21.0",
     "inkfox>=0.1.1",
     "rjieba>=0.1.13",
+    "fastmcp>=2.13.0",
 ]
 
 [[tool.uv.index]]
@@ -110,6 +111,7 @@ ignore = [
     "RUF001", # ambiguous-unicode-character-string
     "RUF002", # ambiguous-unicode-character-docstring
     "RUF003", # ambiguous-unicode-character-comment
+    "PERF203", # try-except-in-loop (我们需要单独处理每个项的错误)
 ]
 
 

requirements.txt

@@ -9,6 +9,7 @@ customtkinter
 dotenv
 faiss-cpu
 fastapi
+fastmcp
 rjieba
 jsonlines
 maim_message

scripts/debug_mcp_tools.py

@@ -0,0 +1,59 @@
+"""
+调试 MCP 工具列表获取
+
+直接测试 MCP 客户端是否能获取工具
+"""
+
+import asyncio
+
+from fastmcp.client import Client, StreamableHttpTransport
+
+
+async def test_direct_connection():
+    """直接连接 MCP 服务器并获取工具列表"""
+    print("=" * 60)
+    print("直接测试 MCP 服务器连接")
+    print("=" * 60)
+
+    url = "http://localhost:8000/mcp"
+    print(f"\n连接到: {url}")
+
+    try:
+        # 创建传输层
+        transport = StreamableHttpTransport(url)
+        print("✓ 传输层创建成功")
+
+        # 创建客户端
+        async with Client(transport) as client:
+            print("✓ 客户端连接成功")
+
+            # 获取工具列表
+            print("\n正在获取工具列表...")
+            tools_result = await client.list_tools()
+
+            print(f"\n获取结果类型: {type(tools_result)}")
+            print(f"结果内容: {tools_result}")
+
+            # 检查是否有 tools 属性
+            if hasattr(tools_result, "tools"):
+                tools = tools_result.tools
+                print(f"\n✓ 找到 tools 属性，包含 {len(tools)} 个工具")
+
+                for i, tool in enumerate(tools, 1):
+                    print(f"\n工具 {i}:")
+                    print(f"  名称: {tool.name}")
+                    print(f"  描述: {tool.description}")
+                    if hasattr(tool, "inputSchema"):
+                        print(f"  参数 Schema: {tool.inputSchema}")
+            else:
+                print("\n✗ 结果中没有 tools 属性")
+                print(f"可用属性: {dir(tools_result)}")
+
+    except Exception as e:
+        print(f"\n✗ 连接失败: {e}")
+        import traceback
+        traceback.print_exc()
+
+
+if __name__ == "__main__":
+    asyncio.run(test_direct_connection())

scripts/simple_mcp_server.py

@@ -0,0 +1,142 @@
+"""
+简单的 MCP 测试服务器
+
+使用 fastmcp 创建一个简单的 MCP 服务器供测试使用。
+
+安装依赖:
+    pip install fastmcp uvicorn
+
+运行服务器:
+    python scripts/simple_mcp_server.py
+
+服务器将在 http://localhost:8000/mcp 提供 MCP 服务
+"""
+
+from fastmcp import FastMCP
+
+# 创建 MCP 服务器实例
+mcp = FastMCP("Demo Server")
+
+
+@mcp.tool()
+def add(a: int, b: int) -> int:
+    """将两个数字相加
+    
+    Args:
+        a: 第一个数字
+        b: 第二个数字
+    
+    Returns:
+        两个数字的和
+    """
+    return a + b
+
+
+@mcp.tool()
+def multiply(a: float, b: float) -> float:
+    """将两个数字相乘
+    
+    Args:
+        a: 第一个数字
+        b: 第二个数字
+    
+    Returns:
+        两个数字的乘积
+    """
+    return a * b
+
+
+@mcp.tool()
+def get_weather(city: str) -> str:
+    """获取指定城市的天气信息(模拟)
+    
+    Args:
+        city: 城市名称
+    
+    Returns:
+        天气信息字符串
+    """
+    # 这是一个模拟实现
+    weather_data = {
+        "beijing": "北京今天晴朗，温度 20°C",
+        "shanghai": "上海今天多云，温度 18°C",
+        "guangzhou": "广州今天有雨，温度 25°C",
+    }
+
+    city_lower = city.lower()
+    return weather_data.get(
+        city_lower,
+        f"{city}的天气信息暂不可用"
+    )
+
+
+@mcp.tool()
+def echo(message: str, repeat: int = 1) -> str:
+    """重复输出一条消息
+    
+    Args:
+        message: 要重复的消息
+        repeat: 重复次数，默认为 1
+    
+    Returns:
+        重复后的消息
+    """
+    return (message + "\n") * repeat
+
+
+@mcp.tool()
+def check_prime(number: int) -> bool:
+    """检查一个数字是否为质数
+    
+    Args:
+        number: 要检查的数字
+    
+    Returns:
+        如果是质数返回 True，否则返回 False
+    """
+    if number < 2:
+        return False
+
+    for i in range(2, int(number ** 0.5) + 1):
+        if number % i == 0:
+            return False
+
+    return True
+
+
+if __name__ == "__main__":
+    print("=" * 60)
+    print("简单 MCP 测试服务器")
+    print("=" * 60)
+    print("\n服务器配置:")
+    print("  - 传输协议: Streamable HTTP")
+    print("  - 地址: http://localhost:8000/mcp")
+    print("\n可用工具:")
+    print("  1. add(a: int, b: int) -> int")
+    print("  2. multiply(a: float, b: float) -> float")
+    print("  3. get_weather(city: str) -> str")
+    print("  4. echo(message: str, repeat: int = 1) -> str")
+    print("  5. check_prime(number: int) -> bool")
+    print("\n配置示例 (config/mcp.json):")
+    print("""
+{
+  "$schema": "./mcp.schema.json",
+  "mcpServers": {
+    "demo_server": {
+      "enabled": true,
+      "transport": {
+        "type": "streamable-http",
+        "url": "http://localhost:8000/mcp"
+      },
+      "timeout": 30
+    }
+  }
+}
+    """)
+    print("=" * 60)
+    print("\n正在启动服务器...")
+    print("请参考 fastmcp 官方文档了解如何运行此服务器。")
+    print("文档: https://github.com/jlowin/fastmcp")
+    print("\n基本命令:")
+    print("  fastmcp run simple_mcp_server:mcp")
+    print("=" * 60)

scripts/test/demo_mcp_server.py

@@ -0,0 +1,15 @@
+from fastmcp import FastMCP
+
+app = FastMCP(
+    name="Demo MCP Server",
+    streamable_http_path="/mcp"
+)
+
+@app.tool()
+async def echo_tool(input: str) -> str:
+    """一个简单的回声工具"""
+    return f"Echo: {input}"
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=8000, transport="streamable-http"
+    )

scripts/test_mcp_integration.py

@@ -0,0 +1,190 @@
+"""
+MCP 集成测试脚本
+
+测试 MCP 客户端连接、工具列表获取和工具调用功能
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+# 添加项目根目录到 Python 路径
+project_root = Path(__file__).parent.parent
+sys.path.insert(0, str(project_root))
+
+from src.common.logger import get_logger
+from src.plugin_system.core.component_registry import ComponentRegistry
+from src.plugin_system.core.mcp_client_manager import MCPClientManager
+
+logger = get_logger("test_mcp_integration")
+
+
+async def test_mcp_client_manager():
+    """测试 MCPClientManager 基本功能"""
+    print("\n" + "="*60)
+    print("测试 1: MCPClientManager 连接和工具列表")
+    print("="*60)
+
+    try:
+        # 初始化 MCP 客户端管理器
+        manager = MCPClientManager()
+        await manager.initialize()
+
+        print("\n✓ MCP 客户端管理器初始化成功")
+        print(f"已连接服务器数量: {len(manager.clients)}")
+
+        # 获取所有工具
+        tools = await manager.get_all_tools()
+        print(f"\n获取到 {len(tools)} 个 MCP 工具:")
+
+        for tool in tools:
+            print(f"\n  工具: {tool}")
+            # 注意: 这里 tool 是字符串形式的工具名称
+            # 如果需要工具详情，需要从其他地方获取
+
+        return manager, tools
+
+    except Exception as e:
+        print(f"\n✗ 测试失败: {e}")
+        logger.exception("MCPClientManager 测试失败")
+        return None, []
+
+
+async def test_tool_call(manager: MCPClientManager, tools):
+    """测试工具调用"""
+    print("\n" + "="*60)
+    print("测试 2: MCP 工具调用")
+    print("="*60)
+
+    if not tools:
+        print("\n⚠ 没有可用的工具进行测试")
+        return
+
+    try:
+        # 工具列表测试已在第一个测试中完成
+        print("\n✓ 工具列表获取成功")
+        print(f"可用工具数量: {len(tools)}")
+
+    except Exception as e:
+        print(f"\n✗ 工具调用测试失败: {e}")
+        logger.exception("工具调用测试失败")
+
+
+async def test_component_registry_integration():
+    """测试 ComponentRegistry 集成"""
+    print("\n" + "="*60)
+    print("测试 3: ComponentRegistry MCP 工具集成")
+    print("="*60)
+
+    try:
+        registry = ComponentRegistry()
+
+        # 加载 MCP 工具
+        await registry.load_mcp_tools()
+
+        # 获取 MCP 工具
+        mcp_tools = registry.get_mcp_tools()
+        print(f"\n✓ ComponentRegistry 加载了 {len(mcp_tools)} 个 MCP 工具")
+
+        for tool in mcp_tools:
+            print(f"\n  工具: {tool.name}")
+            print(f"  描述: {tool.description}")
+            print(f"  参数数量: {len(tool.parameters)}")
+
+            # 测试 is_mcp_tool 方法
+            is_mcp = registry.is_mcp_tool(tool.name)
+            print(f"  is_mcp_tool 检测: {'✓' if is_mcp else '✗'}")
+
+        return mcp_tools
+
+    except Exception as e:
+        print(f"\n✗ ComponentRegistry 集成测试失败: {e}")
+        logger.exception("ComponentRegistry 集成测试失败")
+        return []
+
+
+async def test_tool_execution(mcp_tools):
+    """测试通过适配器执行工具"""
+    print("\n" + "="*60)
+    print("测试 4: MCPToolAdapter 工具执行")
+    print("="*60)
+
+    if not mcp_tools:
+        print("\n⚠ 没有可用的 MCP 工具进行测试")
+        return
+
+    try:
+        # 选择第一个工具测试
+        test_tool = mcp_tools[0]
+        print(f"\n测试工具: {test_tool.name}")
+
+        # 构建测试参数
+        test_args = {}
+        for param_name, param_type, param_desc, is_required, enum_values in test_tool.parameters:
+            if is_required:
+                # 根据类型提供默认值
+                from src.llm_models.payload_content.tool_option import ToolParamType
+
+                if param_type == ToolParamType.STRING:
+                    test_args[param_name] = "test_value"
+                elif param_type == ToolParamType.INTEGER:
+                    test_args[param_name] = 1
+                elif param_type == ToolParamType.FLOAT:
+                    test_args[param_name] = 1.0
+                elif param_type == ToolParamType.BOOLEAN:
+                    test_args[param_name] = True
+
+        print(f"测试参数: {test_args}")
+
+        # 执行工具
+        result = await test_tool.execute(test_args)
+
+        if result:
+            print("\n✓ 工具执行成功")
+            print(f"结果类型: {result.get('type')}")
+            print(f"结果内容: {result.get('content', '')[:200]}...")  # 只显示前200字符
+        else:
+            print("\n✗ 工具执行失败，返回 None")
+
+    except Exception as e:
+        print(f"\n✗ 工具执行测试失败: {e}")
+        logger.exception("工具执行测试失败")
+
+
+async def main():
+    """主测试流程"""
+    print("\n" + "="*60)
+    print("MCP 集成测试")
+    print("="*60)
+
+    try:
+        # 测试 1: MCPClientManager 基本功能
+        manager, tools = await test_mcp_client_manager()
+
+        if manager:
+            # 测试 2: 工具调用
+            await test_tool_call(manager, tools)
+
+            # 测试 3: ComponentRegistry 集成
+            mcp_tools = await test_component_registry_integration()
+
+            # 测试 4: 工具执行
+            await test_tool_execution(mcp_tools)
+
+            # 关闭连接
+            await manager.close()
+            print("\n✓ MCP 客户端连接已关闭")
+
+        print("\n" + "="*60)
+        print("测试完成")
+        print("="*60 + "\n")
+
+    except KeyboardInterrupt:
+        print("\n\n测试被用户中断")
+    except Exception as e:
+        print(f"\n测试过程中发生错误: {e}")
+        logger.exception("测试失败")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/api/statistic_router.py

@@ -219,4 +219,4 @@ async def get_llm_stats(
         raise e
     except Exception as e:
         logger.error(f"获取LLM统计信息失败: {e}")
-        raise HTTPException(status_code=500, detail=str(e))
+        raise HTTPException(status_code=500, detail=str(e))

src/chat/chatter_manager.py

@@ -108,11 +108,6 @@ class ChatterManager:
 
         self.stats["streams_processed"] += 1
         try:
-            # 设置触发用户ID
-            last_message = context.get_last_message()
-            if last_message:
-                context.triggering_user_id = last_message.user_info.user_id
-
             result = await self.instances[stream_id].execute(context)
 
             # 检查执行结果是否真正成功

src/chat/memory_system/memory_system.py

@@ -878,7 +878,7 @@ class MemorySystem:
 
             except Exception as e:
                 logger.warning(f"检索瞬时记忆失败: {e}", exc_info=True)
-            
+
             # 最终截断
             final_memories = final_memories[:effective_limit]
 

src/chat/memory_system/message_collection_processor.py

@@ -72,4 +72,4 @@ class MessageCollectionProcessor:
             "active_buffers": len(self.message_buffers),
             "total_buffered_messages": total_buffered_messages,
             "buffer_capacity_per_chat": self.buffer_size,
-        }
+        }

src/chat/memory_system/message_collection_storage.py

@@ -3,7 +3,6 @@
 专用于存储和检索消息集合，以提供即时上下文。
 """
 
-import asyncio
 import time
 from typing import Any
 
@@ -125,7 +124,7 @@ class MessageCollectionStorage:
             if results and results.get("ids") and results["ids"][0]:
                 for metadata in results["metadatas"][0]:
                     collections.append(MessageCollection.from_dict(metadata))
-            
+
             return collections
         except Exception as e:
             logger.error(f"检索相关消息集合失败: {e}", exc_info=True)
@@ -163,7 +162,7 @@ class MessageCollectionStorage:
 
             # 格式化消息集合为 prompt 上下文
             final_context = "\n\n---\n\n".join(context_parts) + "\n\n---"
-            
+
             logger.info(f"🔗 为查询 '{query_text[:50]}...' 在聊天 '{chat_id}' 中找到 {len(collections)} 个相关消息集合上下文")
             return f"\n{final_context}\n"
 
@@ -192,4 +191,4 @@ class MessageCollectionStorage:
             }
         except Exception as e:
             logger.error(f"获取消息集合存储统计失败: {e}")
-            return {}
+            return {}

src/chat/message_manager/distribution_manager.py

@@ -381,6 +381,11 @@ class StreamLoopManager:
             if cached_messages:
                 logger.info(f"处理开始前刷新缓存消息: stream={stream_id}, 数量={len(cached_messages)}")
 
+            # 设置触发用户ID，以实现回复保护
+            last_message = context.get_last_message()
+            if last_message:
+                context.triggering_user_id = last_message.user_info.user_id
+
             # 创建子任务用于刷新能量（不阻塞主流程）
             energy_task = asyncio.create_task(self._refresh_focus_energy(stream_id))
             child_tasks.add(energy_task)

src/chat/message_receive/chat_stream.py

@@ -156,6 +156,13 @@ class ChatStream:
 
         return instance
 
+    def get_raw_id(self) -> str:
+        """获取原始的、未哈希的聊天流ID字符串"""
+        if self.group_info:
+            return f"{self.platform}:{self.group_info.group_id}:group"
+        else:
+            return f"{self.platform}:{self.user_info.user_id}:private"
+
     def update_active_time(self):
         """更新最后活跃时间"""
         self.last_active_time = time.time()
@@ -256,18 +263,18 @@ class ChatStream:
     def _prepare_additional_config(self, message_info) -> str | None:
         """
         准备 additional_config，将 format_info 嵌入其中
-        
+
         这个方法模仿 storage.py 中的逻辑，确保 DatabaseMessages 中的 additional_config
         包含 format_info，使得 action_modifier 能够正确获取适配器支持的消息类型
-        
+
         Args:
             message_info: BaseMessageInfo 对象
-            
+
         Returns:
             str | None: JSON 字符串格式的 additional_config，如果为空则返回 None
         """
         import orjson
-        
+
         # 首先获取adapter传递的additional_config
         additional_config_data = {}
         if hasattr(message_info, 'additional_config') and message_info.additional_config:

src/chat/message_receive/message.py

@@ -9,10 +9,10 @@ from maim_message import BaseMessageInfo, MessageBase, Seg, UserInfo
 from rich.traceback import install
 
 from src.chat.message_receive.chat_stream import ChatStream
+from src.chat.utils.self_voice_cache import consume_self_voice_text
 from src.chat.utils.utils_image import get_image_manager
 from src.chat.utils.utils_video import get_video_analyzer, is_video_analysis_available
 from src.chat.utils.utils_voice import get_voice_text
-from src.chat.utils.self_voice_cache import consume_self_voice_text
 from src.common.logger import get_logger
 from src.config.config import global_config
 
@@ -212,7 +212,7 @@ class MessageRecv(Message):
                             return f"[语音：{cached_text}]"
                         else:
                             logger.warning("机器人自身语音消息缓存未命中，将回退到标准语音识别。")
-                
+
                 # 标准语音识别流程 (也作为缓存未命中的后备方案)
                 if isinstance(segment.data, str):
                     return await get_voice_text(segment.data)
@@ -239,6 +239,12 @@ class MessageRecv(Message):
                     }
                     """
                 return ""
+            elif segment.type == "file":
+                if isinstance(segment.data, dict):
+                    file_name = segment.data.get('name', '未知文件')
+                    file_size = segment.data.get('size', '未知大小')
+                    return f"[文件：{file_name} ({file_size}字节)]"
+                return "[收到一个文件]"
             elif segment.type == "video":
                 self.is_picid = False
                 self.is_emoji = False
@@ -296,8 +302,8 @@ class MessageRecv(Message):
                 else:
                     return ""
             else:
-                logger.info("未启用视频识别")
-                return "[视频]"
+                logger.warning(f"未知的消息段类型: {segment.type}")
+                return f"[{segment.type} 消息]"
         except Exception as e:
             logger.error(f"处理消息段失败: {e!s}, 类型: {segment.type}, 数据: {segment.data}")
             return f"[处理失败的{segment.type}消息]"
@@ -364,7 +370,7 @@ class MessageRecvS4U(MessageRecv):
                 self.is_picid = False
                 self.is_emoji = False
                 self.is_voice = True
-                
+
                 # 检查消息是否由机器人自己发送
                 # 检查消息是否由机器人自己发送
                 if self.message_info and self.message_info.user_info and str(self.message_info.user_info.user_id) == str(global_config.bot.qq_account):
@@ -433,6 +439,12 @@ class MessageRecvS4U(MessageRecv):
                 self.is_screen = True
                 self.screen_info = segment.data
                 return "屏幕信息"
+            elif segment.type == "file":
+                if isinstance(segment.data, dict):
+                    file_name = segment.data.get('name', '未知文件')
+                    file_size = segment.data.get('size', '未知大小')
+                    return f"[文件：{file_name} ({file_size}字节)]"
+                return "[收到一个文件]"
             elif segment.type == "video":
                 self.is_voice = False
                 self.is_picid = False
@@ -490,8 +502,8 @@ class MessageRecvS4U(MessageRecv):
                 else:
                     return ""
             else:
-                logger.info("未启用视频识别")
-                return "[视频]"
+                logger.warning(f"未知的消息段类型: {segment.type}")
+                return f"[{segment.type} 消息]"
         except Exception as e:
             logger.error(f"处理消息段失败: {e!s}, 类型: {segment.type}, 数据: {segment.data}")
             return f"[处理失败的{segment.type}消息]"

src/chat/replyer/default_generator.py

@@ -796,44 +796,63 @@ class DefaultReplyer:
     async def build_keywords_reaction_prompt(self, target: str | None) -> str:
         """构建关键词反应提示
 
+        该方法根据配置的关键词和正则表达式规则，
+        检查目标消息内容是否触发了任何反应。
+        如果匹配成功，它会生成一个包含所有触发反应的提示字符串，
+        用于指导LLM的回复。
+
         Args:
             target: 目标消息内容
 
         Returns:
-            str: 关键词反应提示字符串
+            str: 关键词反应提示字符串，如果没有触发任何反应则为空字符串
         """
-        # 关键词检测与反应
-        keywords_reaction_prompt = ""
+        if target is None:
+            return ""
+
+        reaction_prompt = ""
         try:
-            # 添加None检查，防止NoneType错误
-            if target is None:
-                return keywords_reaction_prompt
+            current_chat_stream_id_str = self.chat_stream.get_raw_id()
+            # 2. 筛选适用的规则（全局规则 + 特定于当前聊天的规则）
+            applicable_rules = []
+            for rule in global_config.reaction.rules:
+                if rule.chat_stream_id == "" or rule.chat_stream_id == current_chat_stream_id_str:
+                    applicable_rules.append(rule)  # noqa: PERF401
 
-            # 处理关键词规则
-            for rule in global_config.keyword_reaction.keyword_rules:
-                if any(keyword in target for keyword in rule.keywords):
-                    logger.info(f"检测到关键词规则：{rule.keywords}，触发反应：{rule.reaction}")
-                    keywords_reaction_prompt += f"{rule.reaction}，"
+            # 3. 遍历适用规则并执行匹配
+            for rule in applicable_rules:
+                matched = False
+                if rule.rule_type == "keyword":
+                    if any(keyword in target for keyword in rule.patterns):
+                        logger.info(f"检测到关键词规则：{rule.patterns}，触发反应：{rule.reaction}")
+                        reaction_prompt += f"{rule.reaction}，"
+                        matched = True
+
+                elif rule.rule_type == "regex":
+                    for pattern_str in rule.patterns:
+                        try:
+                            pattern = re.compile(pattern_str)
+                            if result := pattern.search(target):
+                                reaction = rule.reaction
+                                # 替换命名捕获组
+                                for name, content in result.groupdict().items():
+                                    reaction = reaction.replace(f"[{name}]", content)
+                                logger.info(f"匹配到正则表达式：{pattern_str}，触发反应：{reaction}")
+                                reaction_prompt += f"{reaction}，"
+                                matched = True
+                                break  # 一个正则规则里只要有一个 pattern 匹配成功即可
+                        except re.error as e:
+                            logger.error(f"正则表达式编译错误: {pattern_str}, 错误信息: {e!s}")
+                            continue
+
+                if matched:
+                    # 如果需要每条消息只触发一个反应规则，可以在这里 break
+                    pass
 
-            # 处理正则表达式规则
-            for rule in global_config.keyword_reaction.regex_rules:
-                for pattern_str in rule.regex:
-                    try:
-                        pattern = re.compile(pattern_str)
-                        if result := pattern.search(target):
-                            reaction = rule.reaction
-                            for name, content in result.groupdict().items():
-                                reaction = reaction.replace(f"[{name}]", content)
-                            logger.info(f"匹配到正则表达式：{pattern_str}，触发反应：{reaction}")
-                            keywords_reaction_prompt += f"{reaction}，"
-                            break
-                    except re.error as e:
-                        logger.error(f"正则表达式编译错误: {pattern_str}, 错误信息: {e!s}")
-                        continue
         except Exception as e:
             logger.error(f"关键词检测与反应时发生异常: {e!s}", exc_info=True)
 
-        return keywords_reaction_prompt
+        return reaction_prompt
 
     async def build_notice_block(self, chat_id: str) -> str:
         """构建notice信息块

src/chat/utils/prompt.py

@@ -303,7 +303,7 @@ class Prompt:
 
     @staticmethod
     def _process_escaped_braces(template) -> str:
-        """预处理模板，将 `\{` 和 `\}` 替换为临时标记."""
+        r"""预处理模板，将 `\{` 和 `\}` 替换为临时标记."""
         if isinstance(template, list):
             template = "\n".join(str(item) for item in template)
         elif not isinstance(template, str):

src/chat/utils/prompt_component_manager.py

@@ -1,6 +1,5 @@
 import asyncio
 import re
-from typing import Type
 
 from src.chat.utils.prompt_params import PromptParameters
 from src.common.logger import get_logger
@@ -21,7 +20,7 @@ class PromptComponentManager:
     3. 提供一个接口，以便在构建核心Prompt时，能够获取并执行所有相关的组件。
     """
 
-    def _get_rules_for(self, target_prompt_name: str) -> list[tuple[InjectionRule, Type[BasePrompt]]]:
+    def _get_rules_for(self, target_prompt_name: str) -> list[tuple[InjectionRule, type[BasePrompt]]]:
         """
         获取指定目标Prompt的所有注入规则及其关联的组件类。
 

src/chat/utils/self_voice_cache.py

@@ -6,15 +6,14 @@
 避免不必要的自我语音识别。
 """
 import hashlib
-from typing import Dict
 
 # 一个简单的内存缓存，用于将机器人自己发送的语音消息映射到其原始文本。
 # 键是语音base64内容的SHA256哈希值。
-_self_voice_cache: Dict[str, str] = {}
+_self_voice_cache: dict[str, str] = {}
 
 def get_voice_key(base64_content: str) -> str:
     """为语音内容生成一个一致的键。"""
-    return hashlib.sha256(base64_content.encode('utf-8')).hexdigest()
+    return hashlib.sha256(base64_content.encode("utf-8")).hexdigest()
 
 def register_self_voice(base64_content: str, text: str):
     """
@@ -39,4 +38,4 @@ def consume_self_voice_text(base64_content: str) -> str | None:
         str | None: 如果找到，则返回原始文本，否则返回None。
     """
     key = get_voice_key(base64_content)
-    return _self_voice_cache.pop(key, None)
+    return _self_voice_cache.pop(key, None)

src/chat/utils/utils.py

@@ -430,19 +430,13 @@ def process_llm_response(text: str, enable_splitter: bool = True, enable_chinese
     if global_config.response_splitter.enable and enable_splitter:
         logger.info(f"回复分割器已启用，模式: {global_config.response_splitter.split_mode}。")
 
-        split_mode = global_config.response_splitter.split_mode
-
-        if split_mode == "llm" and "[SPLIT]" in cleaned_text:
+        if "[SPLIT]" in cleaned_text:
             logger.debug("检测到 [SPLIT] 标记，使用 LLM 自定义分割。")
             split_sentences_raw = cleaned_text.split("[SPLIT]")
             split_sentences = [s.strip() for s in split_sentences_raw if s.strip()]
         else:
-            if split_mode == "llm":
-                logger.debug("未检测到 [SPLIT] 标记，本次不进行分割。")
-                split_sentences = [cleaned_text]
-            else:  # mode == "punctuation"
-                logger.debug("使用基于标点的传统模式进行分割。")
-                split_sentences = split_into_sentences_w_remove_punctuation(cleaned_text)
+            logger.debug("使用基于标点的传统模式进行分割。")
+            split_sentences = split_into_sentences_w_remove_punctuation(cleaned_text)
     else:
         logger.debug("回复分割器已禁用。")
         split_sentences = [cleaned_text]

src/chat/utils/utils_voice.py

@@ -19,10 +19,11 @@ async def get_voice_text(voice_base64: str) -> str:
 
     # 如果选择本地识别
     if asr_provider == "local":
-        from src.plugin_system.apis import tool_api
-        import tempfile
         import base64
         import os
+        import tempfile
+
+        from src.plugin_system.apis import tool_api
 
         local_asr_tool = tool_api.get_tool_instance("local_asr")
         if not local_asr_tool:
@@ -39,8 +40,8 @@ async def get_voice_text(voice_base64: str) -> str:
             text = await local_asr_tool.execute(function_args={"audio_path": audio_path})
             if "失败" in text or "出错" in text or "错误" in text:
                 logger.warning(f"本地语音识别失败: {text}")
-                return f"[语音(本地识别失败)]"
-            
+                return "[语音(本地识别失败)]"
+
             logger.info(f"本地语音识别成功: {text}")
             return f"[语音] {text}"
 

src/config/config.py

@@ -26,7 +26,7 @@ from src.config.official_configs import (
     EmojiConfig,
     ExperimentalConfig,
     ExpressionConfig,
-    KeywordReactionConfig,
+    ReactionConfig,
     LPMMKnowledgeConfig,
     MaimMessageConfig,
     MemoryConfig,
@@ -384,7 +384,7 @@ class Config(ValidatedConfigBase):
     expression: ExpressionConfig = Field(..., description="表达配置")
     memory: MemoryConfig = Field(..., description="记忆配置")
     mood: MoodConfig = Field(..., description="情绪配置")
-    keyword_reaction: KeywordReactionConfig = Field(..., description="关键词反应配置")
+    reaction: ReactionConfig = Field(default_factory=ReactionConfig, description="反应规则配置")
     chinese_typo: ChineseTypoConfig = Field(..., description="中文错别字配置")
     response_post_process: ResponsePostProcessConfig = Field(..., description="响应后处理配置")
     response_splitter: ResponseSplitterConfig = Field(..., description="响应分割配置")

src/config/official_configs.py

@@ -401,32 +401,31 @@ class MoodConfig(ValidatedConfigBase):
     mood_update_threshold: float = Field(default=1.0, description="情绪更新阈值")
 
 
-class KeywordRuleConfig(ValidatedConfigBase):
-    """关键词规则配置类"""
+class ReactionRuleConfig(ValidatedConfigBase):
+    """反应规则配置类"""
 
-    keywords: list[str] = Field(default_factory=lambda: [], description="关键词列表")
-    regex: list[str] = Field(default_factory=lambda: [], description="正则表达式列表")
-    reaction: str = Field(default="", description="反应内容")
+    chat_stream_id: str = Field(default="", description='聊天流ID，格式为 "platform:id:type"，空字符串表示全局')
+    rule_type: Literal["keyword", "regex"] = Field(..., description='规则类型，必须是 "keyword" 或 "regex"')
+    patterns: list[str] = Field(..., description="关键词或正则表达式列表")
+    reaction: str = Field(..., description="触发后的回复内容")
 
     def __post_init__(self):
         import re
 
-        if not self.keywords and not self.regex:
-            raise ValueError("关键词规则必须至少包含keywords或regex中的一个")
-        if not self.reaction:
-            raise ValueError("关键词规则必须包含reaction")
-        for pattern in self.regex:
-            try:
-                re.compile(pattern)
-            except re.error as e:
-                raise ValueError(f"无效的正则表达式 '{pattern}': {e!s}") from e
+        if not self.patterns:
+            raise ValueError("patterns 列表不能为空")
+        if self.rule_type == "regex":
+            for pattern in self.patterns:
+                try:
+                    re.compile(pattern)
+                except re.error as e:
+                    raise ValueError(f"无效的正则表达式 '{pattern}': {e!s}") from e
 
 
-class KeywordReactionConfig(ValidatedConfigBase):
-    """关键词配置类"""
+class ReactionConfig(ValidatedConfigBase):
+    """反应规则系统配置"""
 
-    keyword_rules: list[KeywordRuleConfig] = Field(default_factory=lambda: [], description="关键词规则列表")
-    regex_rules: list[KeywordRuleConfig] = Field(default_factory=lambda: [], description="正则表达式规则列表")
+    rules: list[ReactionRuleConfig] = Field(default_factory=list, description="反应规则列表")
 
 
 class CustomPromptConfig(ValidatedConfigBase):

src/llm_models/utils.py

@@ -14,11 +14,11 @@ from .payload_content.message import Message, MessageBuilder
 logger = get_logger("消息压缩工具")
 
 
-def compress_messages(messages: list[Message], img_target_size: int = 1 * 1024 * 1024) -> list[Message]:
+def compress_messages(messages: list[Message], img_target_size: int = 2 * 1024 * 1024) -> list[Message]:
     """
     压缩消息列表中的图片
     :param messages: 消息列表
-    :param img_target_size: 图片目标大小，默认1MB
+    :param img_target_size: 图片目标大小，默认2MB
     :return: 压缩后的消息列表
     """
 
@@ -29,7 +29,7 @@ def compress_messages(messages: list[Message], img_target_size: int = 1 * 1024 *
         :return: 转换后的图片数据
         """
         try:
-            image = Image.open(image_data)
+            image = Image.open(io.BytesIO(image_data))
 
             if image.format and (image.format.upper() in ["JPEG", "JPG", "PNG", "WEBP"]):
                 # 静态图像，转换为JPEG格式
@@ -50,7 +50,7 @@ def compress_messages(messages: list[Message], img_target_size: int = 1 * 1024 *
         :return: 缩放后的图片数据
         """
         try:
-            image = Image.open(image_data)
+            image = Image.open(io.BytesIO(image_data))
 
             # 原始尺寸
             original_size = (image.width, image.height)

src/main.py

@@ -4,9 +4,10 @@ import signal
 import sys
 import time
 import traceback
+from collections.abc import Callable, Coroutine
 from functools import partial
 from random import choices
-from typing import Any, Callable, Coroutine
+from typing import Any
 
 from maim_message import MessageServer
 from rich.traceback import install
@@ -24,8 +25,8 @@ from src.config.config import global_config
 from src.individuality.individuality import Individuality, get_individuality
 from src.manager.async_task_manager import async_task_manager
 from src.mood.mood_manager import mood_manager
-from src.plugin_system.base.component_types import EventType
 from src.plugin_system.base.base_interest_calculator import BaseInterestCalculator
+from src.plugin_system.base.component_types import EventType
 from src.plugin_system.core.event_manager import event_manager
 from src.plugin_system.core.plugin_manager import plugin_manager
 from src.schedule.monthly_plan_manager import monthly_plan_manager
@@ -418,7 +419,7 @@ MoFox_Bot(第三方修改版)
 
         # 处理所有缓存的事件订阅（插件加载完成后）
         event_manager.process_all_pending_subscriptions()
-        
+
         # 初始化表情管理器
         get_emoji_manager().initialize()
         logger.info("表情包管理器初始化成功")

src/mcp_integration/client_manager.py

@@ -0,0 +1 @@
+

src/plugin_system/apis/send_api.py

@@ -27,6 +27,61 @@
 
 
 """
+from pathlib import Path
+
+
+async def file_to_stream(
+   file_path: str,
+   stream_id: str,
+   file_name: str | None = None,
+   storage_message: bool = True,
+   set_reply: bool = True
+) -> bool:
+   """向指定流发送文件
+
+   Args:
+       file_path: 文件的本地路径
+       stream_id: 聊天流ID
+       file_name: 发送到对方时显示的文件名，如果为 None 则使用原始文件名
+       storage_message: 是否存储消息到数据库
+
+   Returns:
+       bool: 是否发送成功
+   """
+   target_stream = await get_chat_manager().get_stream(stream_id)
+   if not target_stream:
+       logger.error(f"[SendAPI] 未找到聊天流: {stream_id}")
+       return False
+
+   if not file_name:
+       file_name = Path(file_path).name
+   
+   params = {
+       "file": file_path,
+       "name": file_name,
+   }
+
+   action = ""
+   if target_stream.group_info:
+       action = "upload_group_file"
+       params["group_id"] = target_stream.group_info.group_id
+   else:
+       action = "upload_private_file"
+       params["user_id"] = target_stream.user_info.user_id
+   
+   response = await adapter_command_to_stream(
+       action=action,
+       params=params,
+       stream_id=stream_id,
+       timeout=300.0  # 文件上传可能需要更长时间
+   )
+
+   if response.get("status") == "ok":
+       logger.info(f"文件 {file_name} 已成功发送到 {stream_id}")
+       return True
+   else:
+       logger.error(f"文件 {file_name} 发送到 {stream_id} 失败: {response.get('message')}")
+       return False
 
 import asyncio
 import time

src/plugin_system/apis/tool_api.py

@@ -1,4 +1,5 @@
 from typing import Any
+
 from src.common.logger import get_logger
 from src.plugin_system.base.base_tool import BaseTool
 from src.plugin_system.base.component_types import ComponentType
@@ -22,7 +23,7 @@ def get_tool_instance(tool_name: str) -> BaseTool | None:
 
 
 def get_llm_available_tool_definitions() -> list[dict[str, Any]]:
-    """获取LLM可用的工具定义列表
+    """获取LLM可用的工具定义列表（包括 MCP 工具）
 
     Returns:
         list[dict[str, Any]]: 工具定义列表
@@ -31,6 +32,8 @@ def get_llm_available_tool_definitions() -> list[dict[str, Any]]:
 
     llm_available_tools = component_registry.get_llm_available_tools()
     tool_definitions = []
+
+    # 获取常规工具定义
     for tool_name, tool_class in llm_available_tools.items():
         try:
             # 调用类方法 get_tool_definition 获取定义
@@ -38,5 +41,18 @@ def get_llm_available_tool_definitions() -> list[dict[str, Any]]:
             tool_definitions.append(definition)
         except Exception as e:
             logger.error(f"获取工具 {tool_name} 的定义失败: {e}")
+
+    # 获取 MCP 工具定义
+    try:
+        mcp_tools = component_registry.get_mcp_tools()
+        for mcp_tool in mcp_tools:
+            try:
+                definition = mcp_tool.get_tool_definition()
+                tool_definitions.append(definition)
+            except Exception as e:
+                logger.error(f"获取 MCP 工具 {mcp_tool.name} 的定义失败: {e}")
+    except Exception as e:
+        logger.debug(f"获取 MCP 工具列表失败（可能未启用）: {e}")
+
     return tool_definitions
 

src/plugin_system/core/component_registry.py

@@ -87,6 +87,10 @@ class ComponentRegistry:
         self._tool_registry: dict[str, type["BaseTool"]] = {}  # 工具名 -> 工具类
         self._llm_available_tools: dict[str, type["BaseTool"]] = {}  # llm可用的工具名 -> 工具类
 
+        # MCP 工具注册表（运行时动态加载）
+        self._mcp_tools: list["BaseTool"] = []  # MCP 工具适配器实例列表
+        self._mcp_tools_loaded = False  # MCP 工具是否已加载
+
         # EventHandler特定注册表
         self._event_handler_registry: dict[str, type["BaseEventHandler"]] = {}
         """event_handler名 -> event_handler类"""
@@ -891,6 +895,7 @@ class ComponentRegistry:
             "action_components": action_components,
             "command_components": command_components,
             "tool_components": tool_components,
+            "mcp_tools": len(self._mcp_tools),
             "event_handlers": events_handlers,
             "plus_command_components": plus_command_components,
             "chatter_components": chatter_components,
@@ -904,6 +909,34 @@ class ComponentRegistry:
             "enabled_plugins": len([p for p in self._plugins.values() if p.enabled]),
         }
 
+    # === MCP 工具相关方法 ===
+
+    async def load_mcp_tools(self) -> None:
+        """加载 MCP 工具（异步方法）"""
+        if self._mcp_tools_loaded:
+            logger.debug("MCP 工具已加载，跳过")
+            return
+
+        try:
+            from .mcp_tool_adapter import load_mcp_tools_as_adapters
+
+            logger.info("开始加载 MCP 工具...")
+            self._mcp_tools = await load_mcp_tools_as_adapters()
+            self._mcp_tools_loaded = True
+            logger.info(f"MCP 工具加载完成，共 {len(self._mcp_tools)} 个工具")
+        except Exception as e:
+            logger.error(f"加载 MCP 工具失败: {e}")
+            self._mcp_tools = []
+            self._mcp_tools_loaded = True  # 标记为已尝试加载，避免重复尝试
+
+    def get_mcp_tools(self) -> list["BaseTool"]:
+        """获取所有 MCP 工具适配器实例"""
+        return self._mcp_tools.copy()
+
+    def is_mcp_tool(self, tool_name: str) -> bool:
+        """检查工具名是否为 MCP 工具"""
+        return tool_name.startswith("mcp_")
+
     # === 组件移除相关 ===
 
     async def unregister_plugin(self, plugin_name: str) -> bool:

src/plugin_system/core/mcp_client_manager.py

@@ -0,0 +1,266 @@
+"""
+MCP Client Manager
+
+管理多个 MCP (Model Context Protocol) 客户端连接，支持动态加载和工具注册
+"""
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+
+import mcp.types
+from fastmcp.client import Client, StreamableHttpTransport
+
+from src.common.logger import get_logger
+
+logger = get_logger("mcp_client_manager")
+
+
+class MCPServerConfig:
+    """单个 MCP 服务器的配置"""
+
+    def __init__(self, name: str, config: dict[str, Any]):
+        self.name = name
+        self.description = config.get("description", "")
+        self.enabled = config.get("enabled", True)
+        self.transport_config = config["transport"]
+        self.auth_config = config.get("auth")
+        self.timeout = config.get("timeout", 30)
+        self.retry_config = config.get("retry", {"max_retries": 3, "retry_delay": 1})
+
+    def __repr__(self):
+        return f"<MCPServerConfig {self.name} (enabled={self.enabled})>"
+
+
+class MCPClientManager:
+    """
+    MCP 客户端管理器
+
+    负责：
+    1. 从配置文件加载 MCP 服务器配置
+    2. 建立和维护与 MCP 服务器的连接
+    3. 获取可用的工具列表
+    4. 执行工具调用
+    """
+
+    def __init__(self, config_path: str | Path | None = None):
+        """
+        初始化 MCP 客户端管理器
+
+        Args:
+            config_path: mcp.json 配置文件路径，默认为 config/mcp.json
+        """
+        if config_path is None:
+            # 默认配置路径
+
+            config_path = Path(__file__).parent.parent.parent.parent / "config" / "mcp.json"
+
+        self.config_path = Path(config_path)
+        self.servers: dict[str, MCPServerConfig] = {}
+        self.clients: dict[str, Client] = {}
+        self._initialized = False
+        self._lock = asyncio.Lock()
+
+        logger.info(f"MCP 客户端管理器初始化，配置文件: {self.config_path}")
+
+    def load_config(self) -> dict[str, MCPServerConfig]:
+        """
+        从配置文件加载 MCP 服务器配置
+
+        Returns:
+            Dict[str, MCPServerConfig]: 服务器名称 -> 配置对象
+        """
+        if not self.config_path.exists():
+            logger.warning(f"MCP 配置文件不存在: {self.config_path}")
+            return {}
+
+        try:
+            with open(self.config_path, encoding="utf-8") as f:
+                config_data = json.load(f)
+
+            servers = {}
+            mcp_servers = config_data.get("mcpServers", {})
+
+            for server_name, server_config in mcp_servers.items():
+                try:
+                    server = MCPServerConfig(server_name, server_config)
+                    servers[server_name] = server
+                    logger.debug(f"加载 MCP 服务器配置: {server}")
+                except Exception as e:
+                    logger.error(f"加载服务器配置 '{server_name}' 失败: {e}")
+                    continue
+
+            logger.info(f"成功加载 {len(servers)} 个 MCP 服务器配置")
+            return servers
+
+        except json.JSONDecodeError as e:
+            logger.error(f"解析 MCP 配置文件失败: {e}")
+            return {}
+        except Exception as e:
+            logger.error(f"读取 MCP 配置文件失败: {e}")
+            return {}
+
+    async def initialize(self) -> None:
+        """
+        初始化所有启用的 MCP 客户端连接
+
+        这个方法会：
+        1. 加载配置文件
+        2. 为每个启用的服务器创建客户端
+        3. 建立连接并验证
+        """
+        async with self._lock:
+            if self._initialized:
+                logger.debug("MCP 客户端管理器已初始化，跳过")
+                return
+
+            logger.info("开始初始化 MCP 客户端连接...")
+
+            # 加载配置
+            self.servers = self.load_config()
+
+            if not self.servers:
+                logger.warning("没有找到任何 MCP 服务器配置")
+                self._initialized = True
+                return
+
+            # 为每个启用的服务器创建客户端
+            for server_name, server_config in self.servers.items():
+                if not server_config.enabled:
+                    logger.debug(f"服务器 '{server_name}' 未启用，跳过")
+                    continue
+
+                try:
+                    client = await self._create_client(server_config)
+                    self.clients[server_name] = client
+                    logger.info(f"✅ MCP 服务器 '{server_name}' 连接成功")
+                except Exception as e:
+                    logger.error(f"❌ 连接 MCP 服务器 '{server_name}' 失败: {e}")
+                    continue
+
+            self._initialized = True
+            logger.info(f"MCP 客户端管理器初始化完成，成功连接 {len(self.clients)}/{len(self.servers)} 个服务器")
+
+    async def _create_client(self, server_config: MCPServerConfig) -> Client:
+        """
+        根据配置创建 MCP 客户端
+
+        Args:
+            server_config: 服务器配置
+
+        Returns:
+            Client: 已连接的 MCP 客户端
+        """
+        transport_type = server_config.transport_config.get("type", "streamable-http")
+
+        if transport_type == "streamable-http":
+            url = server_config.transport_config["url"]
+            transport = StreamableHttpTransport(url)
+
+            # 设置认证（如果有）
+            if server_config.auth_config:
+                auth_type = server_config.auth_config.get("type")
+                if auth_type == "bearer":
+                    from fastmcp.client.auth import BearerAuth
+
+                    token = server_config.auth_config.get("token", "")
+                    transport._set_auth(BearerAuth(token))
+
+            client = Client(transport, timeout=server_config.timeout)
+
+        elif transport_type == "sse":
+            from fastmcp.client import SSETransport
+
+            url = server_config.transport_config["url"]
+            client = Client(SSETransport(url), timeout=server_config.timeout)
+
+        else:
+            raise ValueError(f"不支持的传输类型: {transport_type}")
+
+        # 进入客户端上下文（建立连接）
+        await client.__aenter__()
+
+        return client
+
+    async def get_all_tools(self) -> dict[str, list[mcp.types.Tool]]:
+        """
+        获取所有 MCP 服务器提供的工具列表
+
+        Returns:
+            Dict[str, List[mcp.types.Tool]]: 服务器名称 -> 工具列表
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        all_tools = {}
+
+        for server_name, client in self.clients.items():
+            try:
+                # fastmcp 的 list_tools() 直接返回 List[Tool]，不是包含 tools 属性的对象
+                tools = await client.list_tools()
+                all_tools[server_name] = tools
+                logger.debug(f"从服务器 '{server_name}' 获取到 {len(tools)} 个工具")
+            except Exception as e:
+                logger.error(f"从服务器 '{server_name}' 获取工具列表失败: {e}")
+                all_tools[server_name] = []
+
+        return all_tools
+
+    async def call_tool(
+        self, server_name: str, tool_name: str, arguments: dict[str, Any] | None = None
+    ) -> Any:
+        """
+        调用指定 MCP 服务器的工具
+
+        Args:
+            server_name: 服务器名称
+            tool_name: 工具名称
+            arguments: 工具参数
+
+        Returns:
+            Any: 工具执行结果（CallToolResult 的兼容类型）
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        if server_name not in self.clients:
+            raise ValueError(f"MCP 服务器 '{server_name}' 未连接")
+
+        client = self.clients[server_name]
+
+        try:
+            logger.debug(f"调用 MCP 工具: {server_name}.{tool_name} | 参数: {arguments}")
+            result = await client.call_tool(tool_name, arguments or {})
+            logger.debug(f"MCP 工具调用成功: {server_name}.{tool_name}")
+            return result
+
+        except Exception as e:
+            logger.error(f"MCP 工具调用失败: {server_name}.{tool_name} | 错误: {e}")
+            raise
+
+    async def close(self) -> None:
+        """关闭所有 MCP 客户端连接"""
+        async with self._lock:
+            if not self._initialized:
+                return
+
+            logger.info("关闭所有 MCP 客户端连接...")
+
+            for server_name, client in self.clients.items():
+                try:
+                    await client.__aexit__(None, None, None)
+                    logger.debug(f"已关闭 MCP 服务器 '{server_name}' 的连接")
+                except Exception as e:
+                    logger.error(f"关闭服务器 '{server_name}' 连接失败: {e}")
+
+            self.clients.clear()
+            self._initialized = False
+            logger.info("所有 MCP 客户端连接已关闭")
+
+    def __repr__(self):
+        return f"<MCPClientManager servers={len(self.servers)} clients={len(self.clients)}>"
+
+
+# 全局单例
+mcp_client_manager = MCPClientManager()

src/plugin_system/core/mcp_tool_adapter.py

@@ -0,0 +1,249 @@
+"""
+MCP Tool Adapter
+
+将 MCP 工具适配为 BaseTool，使其能够被插件系统识别和调用
+"""
+
+from typing import Any, ClassVar
+
+import mcp.types
+
+from src.common.logger import get_logger
+from src.plugin_system.base.base_tool import BaseTool
+from src.plugin_system.base.component_types import ToolParamType
+
+from .mcp_client_manager import mcp_client_manager
+
+logger = get_logger("mcp_tool_adapter")
+
+
+class MCPToolAdapter(BaseTool):
+    """
+    MCP 工具适配器
+
+    将 MCP 协议的工具适配为 BaseTool，使其能够：
+    1. 被插件系统识别和注册
+    2. 被 LLM 调用
+    3. 参与工具缓存机制
+    """
+
+    # 类级别默认值，使用 ClassVar 标注
+    available_for_llm: ClassVar[bool] = True
+
+    def __init__(self, server_name: str, mcp_tool: mcp.types.Tool, plugin_config: dict | None = None):
+        """
+        初始化 MCP 工具适配器
+
+        Args:
+            server_name: MCP 服务器名称
+            mcp_tool: MCP 工具对象
+            plugin_config: 插件配置（可选）
+        """
+        super().__init__(plugin_config)
+
+        self.server_name = server_name
+        self.mcp_tool = mcp_tool
+
+        # 设置实例属性
+        self.name = f"mcp_{server_name}_{mcp_tool.name}"
+        self.description = mcp_tool.description or f"MCP tool from {server_name}"
+
+        # 转换参数定义
+        self.parameters = self._convert_parameters(mcp_tool.inputSchema)
+
+        logger.debug(f"创建 MCP 工具适配器: {self.name}")
+
+    def _convert_parameters(
+        self, input_schema: dict[str, Any] | None
+    ) -> list[tuple[str, ToolParamType, str, bool, list[str] | None]]:
+        """
+        将 MCP 工具的 JSON Schema 参数转换为 BaseTool 参数格式
+
+        Args:
+            input_schema: MCP 工具的 inputSchema (JSON Schema)
+
+        Returns:
+            List[Tuple]: BaseTool 参数格式列表
+        """
+        if not input_schema:
+            return []
+
+        parameters = []
+
+        # JSON Schema 通常有 properties 和 required 字段
+        properties = input_schema.get("properties", {})
+        required_fields = input_schema.get("required", [])
+
+        for param_name, param_def in properties.items():
+            # 获取参数类型
+            param_type_str = param_def.get("type", "string")
+            param_type = self._map_json_type_to_tool_param_type(param_type_str)
+
+            # 获取参数描述
+            param_desc = param_def.get("description", f"Parameter {param_name}")
+
+            # 判断是否必填
+            is_required = param_name in required_fields
+
+            # 获取枚举值（如果有）
+            enum_values = param_def.get("enum")
+
+            parameters.append((param_name, param_type, param_desc, is_required, enum_values))
+
+        return parameters
+
+    @staticmethod
+    def _map_json_type_to_tool_param_type(json_type: str) -> ToolParamType:
+        """
+        将 JSON Schema 类型映射到 ToolParamType
+
+        Args:
+            json_type: JSON Schema 类型字符串
+
+        Returns:
+            ToolParamType: 对应的工具参数类型
+        """
+        type_mapping = {
+            "string": ToolParamType.STRING,
+            "integer": ToolParamType.INTEGER,
+            "number": ToolParamType.FLOAT,
+            "boolean": ToolParamType.BOOLEAN,
+        }
+        return type_mapping.get(json_type, ToolParamType.STRING)
+
+    async def execute(self, function_args: dict[str, Any]) -> dict[str, Any]:
+        """
+        执行 MCP 工具调用
+
+        Args:
+            function_args: 工具调用参数
+
+        Returns:
+            Dict: 工具执行结果
+        """
+        try:
+            logger.debug(f"执行 MCP 工具: {self.name} | 服务器: {self.server_name} | 参数: {function_args}")
+
+            # 移除 llm_called 标记（这是内部使用的）
+            clean_args = {k: v for k, v in function_args.items() if k != "llm_called"}
+
+            # 调用 MCP 客户端管理器执行工具
+            result = await mcp_client_manager.call_tool(
+                server_name=self.server_name, tool_name=self.mcp_tool.name, arguments=clean_args
+            )
+
+            # 解析结果
+            return self._format_result(result)
+
+        except Exception as e:
+            logger.error(f"MCP 工具执行失败: {self.name} | 错误: {e}")
+            return {
+                "type": "error",
+                "content": f"MCP 工具调用失败: {e!s}",
+                "id": self.name,
+            }
+
+    def _format_result(self, result: mcp.types.CallToolResult) -> dict[str, Any]:
+        """
+        格式化 MCP 工具执行结果为标准格式
+
+        Args:
+            result: MCP CallToolResult 对象
+
+        Returns:
+            Dict: 标准化的工具执行结果
+        """
+        # MCP 结果包含 content 列表
+        if not result.content:
+            return {
+                "type": "mcp_result",
+                "content": "",
+                "id": self.name,
+            }
+
+        # 提取所有内容
+        content_parts = []
+        for content_item in result.content:
+            # 根据内容类型提取文本
+            content_type = getattr(content_item, "type", None)
+
+            if content_type == "text":
+                # TextContent 类型
+                text = getattr(content_item, "text", "")
+                content_parts.append(text)
+            elif content_type == "image":
+                # ImageContent 类型
+                data = getattr(content_item, "data", b"")
+                content_parts.append(f"[Image data: {len(data)} bytes]")
+            elif content_type == "audio":
+                # AudioContent 类型
+                data = getattr(content_item, "data", b"")
+                content_parts.append(f"[Audio data: {len(data)} bytes]")
+            else:
+                # 尝试提取 text 或 data 属性
+                text = getattr(content_item, "text", None)
+                if text is not None:
+                    content_parts.append(text)
+                else:
+                    data = getattr(content_item, "data", None)
+                    if data is not None:
+                        data_len = len(data) if hasattr(data, "__len__") else "unknown"
+                        content_parts.append(f"[Binary data: {data_len} bytes]")
+                    else:
+                        content_parts.append(str(content_item))
+
+        return {
+            "type": "mcp_result",
+            "content": "\n".join(content_parts),
+            "id": self.name,
+            "is_error": getattr(result, "isError", False),
+        }
+
+    @classmethod
+    def from_mcp_tool(cls, server_name: str, mcp_tool: mcp.types.Tool) -> "MCPToolAdapter":
+        """
+        从 MCP 工具对象创建适配器实例
+
+        Args:
+            server_name: MCP 服务器名称
+            mcp_tool: MCP 工具对象
+
+        Returns:
+            MCPToolAdapter: 工具适配器实例
+        """
+        return cls(server_name, mcp_tool)
+
+
+async def load_mcp_tools_as_adapters() -> list[MCPToolAdapter]:
+    """
+    加载所有 MCP 工具并转换为适配器
+
+    Returns:
+        List[MCPToolAdapter]: 工具适配器列表
+    """
+    logger.info("开始加载 MCP 工具...")
+
+    # 初始化 MCP 客户端管理器
+    await mcp_client_manager.initialize()
+
+    # 获取所有工具
+    all_tools_dict = await mcp_client_manager.get_all_tools()
+
+    adapters = []
+    total_tools = 0
+
+    for server_name, tools in all_tools_dict.items():
+        logger.debug(f"处理服务器 '{server_name}' 的 {len(tools)} 个工具")
+        total_tools += len(tools)
+
+        for mcp_tool in tools:
+            try:
+                adapter = MCPToolAdapter.from_mcp_tool(server_name, mcp_tool)
+                adapters.append(adapter)
+                logger.debug(f"  ✓ 加载工具: {adapter.name}")
+            except Exception as e:
+                logger.error(f"  ✗ 创建工具适配器失败: {mcp_tool.name} | 错误: {e}")
+                continue
+
+    logger.info(f"MCP 工具加载完成: 成功 {len(adapters)}/{total_tools} 个")
+    return adapters

src/plugin_system/core/tool_use.py

@@ -145,8 +145,7 @@ class ToolExecutor:
         pending_step_two = getattr(self, "_pending_step_two_tools", {})
         if pending_step_two:
             # 添加第二步工具定义
-            for step_two_def in pending_step_two.values():
-                tool_definitions.append(step_two_def)
+            tool_definitions.extend(list(pending_step_two.values()))
 
         return tool_definitions
 
@@ -286,10 +285,33 @@ class ToolExecutor:
             logger.info(
                 f"{self.log_prefix} 正在执行工具: [bold green]{function_name}[/bold green] | 参数: {function_args}"
             )
-            
+
             # 检查是否是MCP工具
-            pass
-            
+            from src.plugin_system.core import component_registry
+
+            if component_registry.is_mcp_tool(function_name):
+                logger.debug(f"{self.log_prefix}识别到 MCP 工具: {function_name}")
+                # 找到对应的 MCP 工具实例
+                mcp_tools = component_registry.get_mcp_tools()
+                mcp_tool = next((t for t in mcp_tools if t.name == function_name), None)
+
+                if mcp_tool:
+                    logger.debug(f"{self.log_prefix}执行 MCP 工具 {function_name}")
+                    result = await mcp_tool.execute(function_args)
+
+                    if result:
+                        logger.debug(f"{self.log_prefix}MCP 工具 {function_name} 执行成功")
+                        return {
+                            "tool_call_id": tool_call.call_id,
+                            "role": "tool",
+                            "name": function_name,
+                            "type": "function",
+                            "content": result.get("content", ""),
+                        }
+                else:
+                    logger.warning(f"{self.log_prefix}未找到 MCP 工具: {function_name}")
+                    return None
+
             function_args["llm_called"] = True  # 标记为LLM调用
 
             # 检查是否是二步工具的第二步调用

src/plugins/built_in/napcat_adapter_plugin/src/recv_handler/__init__.py

@@ -33,7 +33,8 @@ class NoticeType:  # 通知事件
     notify = "notify"
     group_ban = "group_ban"  # 群禁言
     group_msg_emoji_like = "group_msg_emoji_like"  # 群聊表情回复
-
+    group_upload = "group_upload"  # 群文件上传
+ 
     class Notify:
         poke = "poke"  # 戳一戳
         input_status = "input_status"  # 正在输入
@@ -59,6 +60,7 @@ class RealMessageType:  # 实际消息分类
     forward = "forward"  # 转发消息
     node = "node"  # 转发消息节点
     json = "json"  # json消息
+    file = "file"  # 文件
 
 
 class MessageSentType:

src/plugins/built_in/napcat_adapter_plugin/src/recv_handler/message_handler.py

@@ -178,10 +178,6 @@ class MessageHandler:
         message_time: float = time.time()  # 应可乐要求，现在是float了
 
         template_info: TemplateInfo = None  # 模板信息，暂时为空，等待启用
-        format_info: FormatInfo = FormatInfo(
-            content_format=["text", "image", "emoji", "voice"],
-            accept_format=ACCEPT_FORMAT,
-        )  # 格式化信息
         if message_type == MessageType.private:
             sub_type = raw_message.get("sub_type")
             if sub_type == MessageType.Private.friend:
@@ -275,6 +271,25 @@ class MessageHandler:
                 logger.warning(f"群聊消息类型 {sub_type} 不支持")
                 return None
 
+        # 处理实际信息
+        if not raw_message.get("message"):
+            logger.warning("原始消息内容为空")
+            return None
+
+        # 获取Seg列表
+        seg_message: List[Seg] = await self.handle_real_message(raw_message)
+        if not seg_message:
+            logger.warning("处理后消息内容为空")
+            return None
+
+        # 动态生成 content_format
+        content_formats = sorted(list(set(seg.type for seg in seg_message)))
+        logger.debug(f"动态生成 content_format: {content_formats}")
+        format_info: FormatInfo = FormatInfo(
+            content_format=content_formats,
+            accept_format=ACCEPT_FORMAT,
+        )
+
         additional_config: dict = {}
         if config_api.get_plugin_config(self.plugin_config, "voice.use_tts"):
             additional_config["allow_tts"] = True
@@ -291,17 +306,6 @@ class MessageHandler:
             additional_config=additional_config,
         )
 
-        # 处理实际信息
-        if not raw_message.get("message"):
-            logger.warning("原始消息内容为空")
-            return None
-
-        # 获取Seg列表
-        seg_message: List[Seg] = await self.handle_real_message(raw_message)
-        if not seg_message:
-            logger.warning("处理后消息内容为空")
-            return None
-
         # 消息缓冲功能已移除，直接处理消息
 
         logger.debug(f"准备发送消息到MoFox-Bot，消息段数量: {len(seg_message)}")
@@ -482,6 +486,13 @@ class MessageHandler:
                         seg_message.append(ret_seg)
                     else:
                         logger.warning("json处理失败")
+                case RealMessageType.file:
+                    ret_seg = await self.handle_file_message(sub_message)
+                    if ret_seg:
+                        # NapcatEvent doesn't have a FILE event yet, so we won't trigger one for now.
+                        seg_message.append(ret_seg)
+                    else:
+                        logger.warning("file处理失败")
                 case _:
                     logger.warning(f"未知消息类型: {sub_message_type}")
 
@@ -773,8 +784,19 @@ class MessageHandler:
             return Seg(type="json", data=json.dumps(message_data))
 
         try:
+            # 尝试将json_data解析为Python对象
             nested_data = json.loads(json_data)
 
+            # 检查是否是机器人自己上传文件的回声
+            if self._is_file_upload_echo(nested_data):
+                logger.info("检测到机器人发送文件的回声消息，将作为文件消息处理")
+                # 从回声消息中提取文件信息
+                file_info = self._extract_file_info_from_echo(nested_data)
+                if file_info:
+                    # 构建一个与普通文件消息格式相同的字典
+                    file_message_dict = {"type": "file", "data": file_info}
+                    return await self.handle_file_message(file_message_dict)
+
             # 检查是否是QQ小程序分享消息
             if "app" in nested_data and "com.tencent.miniapp" in str(nested_data.get("app", "")):
                 logger.debug("检测到QQ小程序分享消息，开始提取信息")
@@ -888,13 +910,88 @@ class MessageHandler:
             # 如果没有提取到关键信息，返回None
             return None
 
-        except json.JSONDecodeError as e:
-            logger.error(f"解析JSON消息失败: {e}")
+        except json.JSONDecodeError:
+            # 如果解析失败，我们假设它不是我们关心的任何一种结构化JSON，
+            # 而是普通的文本或者无法解析的格式。
+            logger.debug(f"无法将data字段解析为JSON: {json_data}")
             return None
         except Exception as e:
-            logger.error(f"处理JSON消息时出错: {e}")
+            logger.error(f"处理JSON消息时发生未知错误: {e}")
             return None
 
+    async def handle_file_message(self, raw_message: dict) -> Seg | None:
+        """
+        处理文件消息
+        Parameters:
+            raw_message: dict: 原始消息
+        Returns:
+            seg_data: Seg: 处理后的消息段
+        """
+        message_data: dict = raw_message.get("data")
+        if not message_data:
+            logger.warning("文件消息缺少 data 字段")
+            return None
+
+        # 提取文件信息
+        file_name = message_data.get("file")
+        file_size = message_data.get("file_size")
+        file_id = message_data.get("file_id")
+
+        logger.info(f"收到文件消息: name={file_name}, size={file_size}, id={file_id}")
+
+        # 将文件信息打包成字典
+        file_data = {
+            "name": file_name,
+            "size": file_size,
+            "id": file_id,
+        }
+
+        return Seg(type="file", data=file_data)
+
+    def _is_file_upload_echo(self, nested_data: Any) -> bool:
+        """检查一个JSON对象是否是机器人自己上传文件的回声消息"""
+        if not isinstance(nested_data, dict):
+            return False
+
+        # 检查 'app' 和 'meta' 字段是否存在
+        if "app" not in nested_data or "meta" not in nested_data:
+            return False
+
+        # 检查 'app' 字段是否包含 'com.tencent.miniapp'
+        if "com.tencent.miniapp" not in str(nested_data.get("app", "")):
+            return False
+
+        # 检查 'meta' 内部的 'detail_1' 的 'busi_id' 是否为 '1014'
+        meta = nested_data.get("meta", {})
+        detail_1 = meta.get("detail_1", {})
+        if detail_1.get("busi_id") == "1014":
+            return True
+
+        return False
+
+    def _extract_file_info_from_echo(self, nested_data: dict) -> Optional[dict]:
+        """从文件上传的回声消息中提取文件信息"""
+        try:
+            meta = nested_data.get("meta", {})
+            detail_1 = meta.get("detail_1", {})
+            
+            # 文件名在 'desc' 字段
+            file_name = detail_1.get("desc")
+            
+            # 文件大小在 'summary' 字段，格式为 "大小：1.7MB"
+            summary = detail_1.get("summary", "")
+            file_size_str = summary.replace("大小：", "").strip() # 移除前缀和空格
+            
+            # QQ API有时返回的大小不标准，这里我们只提取它给的字符串
+            # 实际大小已经由Napcat在发送时记录，这里主要是为了保持格式一致
+            
+            if file_name and file_size_str:
+                return {"file": file_name, "file_size": file_size_str, "file_id": None} # file_id在回声中不可用
+        except Exception as e:
+            logger.error(f"从文件回声中提取信息失败: {e}")
+            
+        return None
+        
     async def handle_rps_message(self, raw_message: dict) -> Seg:
         message_data: dict = raw_message.get("data", {})
         res = message_data.get("result", "")

src/plugins/built_in/napcat_adapter_plugin/src/recv_handler/notice_handler.py

@@ -100,6 +100,7 @@ class NoticeHandler:
         # message_time: int = raw_message.get("time")
         message_time: float = time.time()  # 应可乐要求，现在是float了
 
+        self_id = raw_message.get("self_id")
         group_id = raw_message.get("group_id")
         user_id = raw_message.get("user_id")
         target_id = raw_message.get("target_id")
@@ -159,6 +160,14 @@ class NoticeHandler:
                         system_notice = True
                     case _:
                         logger.warning(f"不支持的group_ban类型: {notice_type}.{sub_type}")
+            case NoticeType.group_upload:
+                logger.info("群文件上传")
+                if user_id == self_id:
+                    logger.info("检测到机器人自己上传文件，忽略此通知")
+                    return None
+                if not await message_handler.check_allow_to_chat(user_id, group_id, False, False):
+                    return None
+                handled_message, user_info = await self.handle_group_upload_notify(raw_message, group_id, user_id, self_id)
             case _:
                 logger.warning(f"不支持的notice类型: {notice_type}")
                 return None
@@ -211,6 +220,9 @@ class NoticeHandler:
         elif notice_type == NoticeType.group_msg_emoji_like:
             notice_config["notice_type"] = "emoji_like"
             notice_config["is_notice"] = True  # 表情回复也是notice
+        elif notice_type == NoticeType.group_upload:
+            notice_config["notice_type"] = "group_upload"
+            notice_config["is_notice"] = True  # 文件上传也是notice
         
         message_info: BaseMessageInfo = BaseMessageInfo(
             platform=config_api.get_plugin_config(self.plugin_config, "maibot_server.platform_name", "qq"),
@@ -373,6 +385,38 @@ class NoticeHandler:
                         )     
         seg_data = Seg(type="text",data=f"{user_name}使用Emoji表情{QQ_FACE.get(like_emoji_id,"")}回复了你的消息[{target_message_text}]")
         return seg_data, user_info
+
+    async def handle_group_upload_notify(self, raw_message: dict, group_id: int, user_id: int, self_id: int):
+        if not group_id:
+            logger.error("群ID不能为空，无法处理群文件上传通知")
+            return None, None
+
+        user_qq_info: dict = await get_member_info(self.get_server_connection(), group_id, user_id)
+        if user_qq_info:
+            user_name = user_qq_info.get("nickname")
+            user_cardname = user_qq_info.get("card")
+        else:
+            user_name = "QQ用户"
+            user_cardname = "QQ用户"
+            logger.debug("无法获取上传文件的用户昵称")
+
+        file_info = raw_message.get("file")
+        if not file_info:
+            logger.error("群文件上传通知中缺少文件信息")
+            return None, None
+
+        user_info: UserInfo = UserInfo(
+            platform=config_api.get_plugin_config(self.plugin_config, "maibot_server.platform_name", "qq"),
+            user_id=user_id,
+            user_nickname=user_name,
+            user_cardname=user_cardname,
+        )
+
+        file_name = file_info.get("name", "未知文件")
+        file_size = file_info.get("size", 0)
+        
+        seg_data = Seg(type="text", data=f"{user_name} 上传了文件: {file_name} (大小: {file_size} 字节)")
+        return seg_data, user_info
     
     async def handle_ban_notify(self, raw_message: dict, group_id: int) -> Tuple[Seg, UserInfo] | Tuple[None, None]:
         if not group_id:

src/plugins/built_in/napcat_adapter_plugin/src/send_handler.py

@@ -274,6 +274,9 @@ class SendHandler:
             new_payload = self.build_payload(payload, self.handle_videourl_message(video_url), False)
         elif seg.type == "file":
             file_path = seg.data
+            file_path = seg.data
+            if isinstance(file_path, dict):
+                file_path = file_path.get("file", "")
             new_payload = self.build_payload(payload, self.handle_file_message(file_path), False)
         return new_payload
 
@@ -411,6 +414,10 @@ class SendHandler:
 
     def handle_file_message(self, file_path: str) -> dict:
         """处理文件消息"""
+        if not file_path:
+            logger.error("文件路径为空")
+            return {}
+
         return {
             "type": "file",
             "data": {"file": f"file://{file_path}"},

src/plugins/built_in/stt_whisper_plugin/__init__.py

@@ -6,4 +6,4 @@ __plugin_meta__ = PluginMetadata(
     usage="在 bot_config.toml 中将 asr_provider 设置为 'local' 即可启用",
     version="0.1.0",
     author="Elysia",
-)
+)

src/plugins/built_in/stt_whisper_plugin/plugin.py

@@ -1,9 +1,4 @@
 import asyncio
-import os
-import tempfile
-from typing import Any
-from pathlib import Path
-import toml
 
 import whisper
 
@@ -40,7 +35,7 @@ class LocalASRTool(BaseTool):
                 model_size = plugin_config.get("whisper", {}).get("model_size", "tiny")
                 device = plugin_config.get("whisper", {}).get("device", "cpu")
                 logger.info(f"正在预加载 Whisper ASR 模型: {model_size} ({device})")
-                
+
                 loop = asyncio.get_running_loop()
                 _whisper_model = await loop.run_in_executor(
                     None, whisper.load_model, model_size, device
@@ -61,10 +56,10 @@ class LocalASRTool(BaseTool):
         # 增强的等待逻辑：只要模型还没准备好，就一直等待后台加载任务完成
         while _is_loading:
             await asyncio.sleep(0.2)
-        
+
         if _whisper_model is None:
             return "Whisper 模型加载失败，无法识别语音。"
-        
+
         try:
             logger.info(f"开始使用 Whisper 识别音频: {audio_path}")
             loop = asyncio.get_running_loop()
@@ -110,6 +105,6 @@ class STTWhisperPlugin(BasePlugin):
                 ), LocalASRTool)]
         except Exception as e:
             logger.error(f"检查 ASR provider 配置时出错: {e}")
-        
+
         logger.debug("ASR provider is not 'local', whisper plugin's tool is disabled.")
         return []

src/plugins/built_in/tts_voice_plugin/actions/tts_action.py

@@ -6,9 +6,9 @@ from pathlib import Path
 
 import toml
 
+from src.chat.utils.self_voice_cache import register_self_voice
 from src.common.logger import get_logger
 from src.plugin_system.base.base_action import BaseAction, ChatMode
-from src.chat.utils.self_voice_cache import register_self_voice
 
 from ..services.manager import get_service
 

src/plugins/built_in/web_search_tool/__init__.py

@@ -15,7 +15,7 @@ __plugin_meta__ = PluginMetadata(
         "is_built_in": True,
     },
     # Python包依赖列表
-    python_dependencies = [  # noqa: RUF012
+    python_dependencies = [
         PythonDependency(package_name="asyncddgs", description="异步DuckDuckGo搜索库", optional=False),
         PythonDependency(
             package_name="exa_py",

template/bot_config_template.toml

@@ -279,12 +279,6 @@ max_frequency_bonus = 10.0 # 最大激活频率奖励天数
 # 休眠机制
 dormant_threshold_days = 90 # 休眠状态判定天数（超过此天数未访问的记忆进入休眠状态）
 
-# 统一存储配置 (已弃用 - 请使用Vector DB配置)
-# DEPRECATED: unified_storage_path = "data/unified_memory"
-# DEPRECATED: unified_storage_cache_limit = 10000
-# DEPRECATED: unified_storage_auto_save_interval = 50
-# DEPRECATED: unified_storage_enable_compression = true
-
 # Vector DB存储配置 (新增 - 替代JSON存储)
 enable_vector_memory_storage = true # 启用Vector DB存储
 enable_llm_instant_memory = true # 启用基于LLM的瞬时记忆
@@ -336,22 +330,36 @@ qa_ppr_damping = 0.8 # PPR阻尼系数
 qa_res_top_k = 3 # 最终提供的文段TopK
 embedding_dimension = 1024 # 嵌入向量维度,应该与模型的输出维度一致
 
-# keyword_rules 用于设置关键词触发的额外回复知识
-# 添加新规则方法：在 keyword_rules 数组中增加一项，格式如下：
-# { keywords = ["关键词1", "关键词2"], reaction = "触发这些关键词时的回复内容" }
-# 例如，添加一个新规则：当检测到“你好”或“hello”时回复“你好，有什么可以帮你？”
-# { keywords = ["你好", "hello"], reaction = "你好，有什么可以帮你？" }
-[keyword_reaction]
-keyword_rules = [
-    { keywords = ["人机", "bot", "机器", "入机", "robot", "机器人", "ai", "AI"], reaction = "有人说你是人机或者机器人，否定这一事实，攻击或反问他。注意，只有认为你是机器人才回复，否则不要否认" },
-    { keywords = ["测试关键词回复", "test"], reaction = "回答测试成功" },
-    #{ keywords = ["你好", "hello"], reaction = "你好，有什么可以帮你？" }    
-    # 在此处添加更多规则，格式同上
-]
+# --- 反应规则系统 ---
+# 在这里，您可以定义一系列基于关键词或正则表达式的自动回复规则。
+# 每条规则都是一个独立的 [[reaction.rules]] 块。
 
-regex_rules = [
-    { regex = ["^(?P<n>\\S{1,20})是这样的$"], reaction = "请按照以下模板造句：[n]是这样的，xx只要xx就可以，可是[n]要考虑的事情就很多了，比如什么时候xx，什么时候xx，什么时候xx。（请自由发挥替换xx部分，只需保持句式结构，同时表达一种将[n]过度重视的反讽意味）" }
-]
+# chat_stream_id: 聊天流ID (格式 "platform:id:type")。
+#                 用于指定此规则仅在哪个聊天中生效。
+#                 如果留空 ("")，则为全局规则，对所有聊天生效。
+# rule_type:      规则类型，必须是 "keyword" 或 "regex"。
+#                 "keyword": 表示本条规则使用关键词匹配。
+#                 "regex":   表示本条规则使用正则表达式匹配。
+# patterns:       一个字符串列表，根据 rule_type 的不同，这里填写关键词或正则表达式。
+# reaction:       触发规则后，机器人发送的回复内容。
+
+[[reaction.rules]]
+chat_stream_id = ""
+rule_type = "keyword"
+patterns = ["人机", "bot", "机器", "入机", "robot", "机器人", "ai", "AI"]
+reaction = "有人说你是人机或者机器人，否定这一事实，攻击或反问他。注意，只有认为你是机器人才回复，否则不要否认"
+
+[[reaction.rules]]
+chat_stream_id = ""
+rule_type = "keyword"
+patterns = ["测试关键词回复", "test"]
+reaction = "回答测试成功"
+
+[[reaction.rules]]
+chat_stream_id = ""
+rule_type = "regex"
+patterns = ["^(?P<n>\\S{1,20})是这样的$"]
+reaction = "请按照以下模板造句：[n]是这样的，xx只要xx就可以，可是[n]要考虑的事情就很多了，比如什么时候xx，什么时候xx，什么时候xx。（请自由发挥替换xx部分，只需保持句式结构，同时表达一种将[n]过度重视的反讽意味）"
 
 # 可以自定义部分提示词
 [custom_prompt]