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

src/chat/memory_system/memory_system.py

@@ -1111,7 +1111,6 @@ class MemorySystem:
                 from src.chat.message_receive.chat_stream import get_chat_manager
 
                 chat_manager = get_chat_manager()
-                # ChatManager.get_stream 是异步方法，需要 await，否则会产生 "coroutine was never awaited" 警告
                 chat_stream = await chat_manager.get_stream(stream_id)
                 if chat_stream and hasattr(chat_stream, "context_manager"):
                     history_limit = self._determine_history_limit(context)

src/chat/planner_actions/action_manager.py

@@ -615,6 +615,7 @@ class ChatterActionManager:
         """禁用批量存储模式"""
         self._batch_storage_enabled = False
         self._current_chat_id = None
+        self._pending_actions = []  # 清空队列
         logger.debug("已禁用批量存储模式")
 
     def add_action_to_batch(self, action_name: str, action_data: dict, thinking_id: str = "",

src/chat/replyer/replyer_manager.py

@@ -37,7 +37,6 @@ class ReplyerManager:
         target_stream = chat_stream
         if not target_stream:
             if chat_manager := get_chat_manager():
-                # get_stream 为异步，需要等待
                 target_stream = await chat_manager.get_stream(stream_id)
 
         if not target_stream:

src/config/api_ada_configs.py

@@ -12,8 +12,8 @@ class APIProvider(ValidatedConfigBase):
     name: str = Field(..., min_length=1, description="API提供商名称")
     base_url: str = Field(..., description="API基础URL")
     api_key: str | list[str] = Field(..., min_length=1, description="API密钥，支持单个密钥或密钥列表轮询")
-    client_type: Literal["openai", "gemini", "aiohttp_gemini"] = Field(
-        default="openai", description="客户端类型（如openai/google等，默认为openai）"
+    client_type: Literal["openai", "gemini", "aiohttp_gemini", "mcp_sse"] = Field(
+        default="openai", description="客户端类型（如openai/google/mcp_sse等，默认为openai）"
     )
     max_retry: int = Field(default=2, ge=0, description="最大重试次数（单个模型API调用失败，最多重试的次数）")
     timeout: int = Field(

src/main.py

@@ -339,6 +339,16 @@ MoFox_Bot(第三方修改版)
 
         # 处理所有缓存的事件订阅（插件加载完成后）
         event_manager.process_all_pending_subscriptions()
+        
+        # 初始化MCP工具提供器
+        try:
+            mcp_config = global_config.get("mcp_servers", [])
+            if mcp_config:
+                from src.plugin_system.utils.mcp_tool_provider import mcp_tool_provider
+                await mcp_tool_provider.initialize(mcp_config)
+                logger.info("MCP工具提供器初始化成功")
+        except Exception as e:
+            logger.info(f"MCP工具提供器未配置或初始化失败: {e}")
 
         # 初始化表情管理器
         get_emoji_manager().initialize()

src/plugin_system/apis/generator_api.py

@@ -9,17 +9,18 @@
 """
 
 import traceback
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from rich.traceback import install
 
 from src.chat.message_receive.chat_stream import ChatStream
-from src.chat.replyer.default_generator import DefaultReplyer
-from src.chat.replyer.replyer_manager import replyer_manager
 from src.chat.utils.utils import process_llm_response
 from src.common.logger import get_logger
 from src.plugin_system.base.component_types import ActionInfo
 
+if TYPE_CHECKING:
+    from src.chat.replyer.default_generator import DefaultReplyer
+
 install(extra_lines=3)
 
 # 日志记录器
@@ -35,7 +36,7 @@ async def get_replyer(
     chat_stream: ChatStream | None = None,
     chat_id: str | None = None,
     request_type: str = "replyer",
-) -> DefaultReplyer | None:
+) -> Any | None:
     """获取回复器对象
 
     优先使用chat_stream，如果没有则使用chat_id直接查找。
@@ -56,6 +57,8 @@ async def get_replyer(
         raise ValueError("chat_stream 和 chat_id 不可均为空")
     try:
         logger.debug(f"[GeneratorAPI] 正在获取回复器，chat_id: {chat_id}, chat_stream: {'有' if chat_stream else '无'}")
+        # 动态导入避免循环依赖
+        from src.chat.replyer.replyer_manager import replyer_manager
         return await replyer_manager.get_replyer(
             chat_stream=chat_stream,
             chat_id=chat_id,

src/plugin_system/apis/tool_api.py

@@ -17,7 +17,12 @@ def get_tool_instance(tool_name: str) -> BaseTool | None:
         plugin_config = None
 
     tool_class: type[BaseTool] = component_registry.get_component_class(tool_name, ComponentType.TOOL)  # type: ignore
-    return tool_class(plugin_config) if tool_class else None
+    if tool_class:
+        return tool_class(plugin_config)
+    
+    # 如果不是常规工具，检查是否是MCP工具
+    # MCP工具不需要返回实例，会在execute_tool_call中特殊处理
+    return None
 
 
 def get_llm_available_tool_definitions():
@@ -29,4 +34,16 @@ def get_llm_available_tool_definitions():
     from src.plugin_system.core import component_registry
 
     llm_available_tools = component_registry.get_llm_available_tools()
-    return [(name, tool_class.get_tool_definition()) for name, tool_class in llm_available_tools.items()]
+    tool_definitions = [(name, tool_class.get_tool_definition()) for name, tool_class in llm_available_tools.items()]
+    
+    # 添加MCP工具
+    try:
+        from src.plugin_system.utils.mcp_tool_provider import mcp_tool_provider
+        mcp_tools = mcp_tool_provider.get_mcp_tool_definitions()
+        tool_definitions.extend(mcp_tools)
+        if mcp_tools:
+            logger.debug(f"已添加 {len(mcp_tools)} 个MCP工具到可用工具列表")
+    except Exception as e:
+        logger.debug(f"获取MCP工具失败（可能未配置）: {e}")
+    
+    return tool_definitions

src/plugin_system/core/tool_use.py

@@ -279,6 +279,23 @@ class ToolExecutor:
             logger.info(
                 f"{self.log_prefix} 正在执行工具: [bold green]{function_name}[/bold green] | 参数: {function_args}"
             )
+            
+            # 检查是否是MCP工具
+            try:
+                from src.plugin_system.utils.mcp_tool_provider import mcp_tool_provider
+                if function_name in mcp_tool_provider.mcp_tools:
+                    logger.info(f"{self.log_prefix}执行MCP工具: {function_name}")
+                    result = await mcp_tool_provider.call_mcp_tool(function_name, function_args)
+                    return {
+                        "tool_call_id": tool_call.call_id,
+                        "role": "tool",
+                        "name": function_name,
+                        "type": "function",
+                        "content": result.get("content", ""),
+                    }
+            except Exception as e:
+                logger.debug(f"检查MCP工具时出错: {e}")
+            
             function_args["llm_called"] = True  # 标记为LLM调用
 
             # 检查是否是二步工具的第二步调用

src/plugin_system/utils/mcp_connector.py

@@ -0,0 +1,235 @@
+"""
+MCP (Model Context Protocol) 连接器
+负责连接MCP服务器，获取和执行工具
+"""
+
+import asyncio
+from typing import Any
+
+import aiohttp
+import orjson
+
+from src.common.logger import get_logger
+
+logger = get_logger("MCP连接器")
+
+
+class MCPConnector:
+    """MCP服务器连接器"""
+
+    def __init__(self, server_url: str, api_key: str | None = None, timeout: int = 30):
+        """
+        初始化MCP连接器
+
+        Args:
+            server_url: MCP服务器URL
+            api_key: API密钥（可选）
+            timeout: 超时时间（秒）
+        """
+        self.server_url = server_url.rstrip("/")
+        self.api_key = api_key
+        self.timeout = timeout
+        self._session: aiohttp.ClientSession | None = None
+        self._tools_cache: dict[str, dict[str, Any]] = {}
+        self._cache_timestamp: float = 0
+        self._cache_ttl: int = 300  # 工具列表缓存5分钟
+
+    async def _get_session(self) -> aiohttp.ClientSession:
+        """获取或创建aiohttp会话"""
+        if self._session is None or self._session.closed:
+            timeout = aiohttp.ClientTimeout(total=self.timeout)
+            self._session = aiohttp.ClientSession(timeout=timeout)
+        return self._session
+
+    async def close(self):
+        """关闭连接"""
+        if self._session and not self._session.closed:
+            await self._session.close()
+
+    def _build_headers(self) -> dict[str, str]:
+        """构建请求头"""
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        }
+        if self.api_key:
+            headers["Authorization"] = f"Bearer {self.api_key}"
+        return headers
+
+    async def list_tools(self, force_refresh: bool = False) -> dict[str, dict[str, Any]]:
+        """
+        获取MCP服务器提供的工具列表
+
+        Args:
+            force_refresh: 是否强制刷新缓存
+
+        Returns:
+            Dict[str, Dict]: 工具字典，key为工具名，value为工具定义
+        """
+        import time
+
+        # 检查缓存
+        if not force_refresh and self._tools_cache and (time.time() - self._cache_timestamp) < self._cache_ttl:
+            logger.debug("使用缓存的MCP工具列表")
+            return self._tools_cache
+
+        logger.info(f"正在从MCP服务器获取工具列表: {self.server_url}")
+
+        try:
+            session = await self._get_session()
+            url = f"{self.server_url}/tools/list"
+
+            async with session.post(url, headers=self._build_headers(), json={}) as response:
+                if response.status != 200:
+                    error_text = await response.text()
+                    logger.error(f"获取MCP工具列表失败: HTTP {response.status} - {error_text}")
+                    return {}
+
+                data = await response.json()
+
+                # 解析工具列表
+                tools = {}
+                tool_list = data.get("tools", [])
+
+                for tool_def in tool_list:
+                    tool_name = tool_def.get("name")
+                    if not tool_name:
+                        continue
+
+                    tools[tool_name] = {
+                        "name": tool_name,
+                        "description": tool_def.get("description", ""),
+                        "input_schema": tool_def.get("inputSchema", {}),
+                    }
+
+                logger.info(f"成功获取 {len(tools)} 个MCP工具")
+                self._tools_cache = tools
+                self._cache_timestamp = time.time()
+
+                return tools
+
+        except aiohttp.ClientError as e:
+            logger.error(f"连接MCP服务器失败: {e}")
+            return {}
+        except Exception as e:
+            logger.error(f"获取MCP工具列表时发生错误: {e}")
+            return {}
+
+    async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+        """
+        调用MCP服务器上的工具
+
+        Args:
+            tool_name: 工具名称
+            arguments: 工具参数
+
+        Returns:
+            Dict: 工具执行结果
+        """
+        logger.info(f"调用MCP工具: {tool_name}")
+        logger.debug(f"工具参数: {arguments}")
+
+        try:
+            session = await self._get_session()
+            url = f"{self.server_url}/tools/call"
+
+            payload = {"name": tool_name, "arguments": arguments}
+
+            async with session.post(url, headers=self._build_headers(), json=payload) as response:
+                if response.status != 200:
+                    error_text = await response.text()
+                    logger.error(f"MCP工具调用失败: HTTP {response.status} - {error_text}")
+                    return {
+                        "success": False,
+                        "error": f"HTTP {response.status}: {error_text}",
+                        "content": f"调用MCP工具 {tool_name} 失败",
+                    }
+
+                result = await response.json()
+
+                # 提取内容
+                content = result.get("content", [])
+                if isinstance(content, list) and len(content) > 0:
+                    # MCP返回的是content数组
+                    text_content = []
+                    for item in content:
+                        if isinstance(item, dict):
+                            if item.get("type") == "text":
+                                text_content.append(item.get("text", ""))
+                        else:
+                            text_content.append(str(item))
+
+                    result_text = "\n".join(text_content) if text_content else str(content)
+                else:
+                    result_text = str(content)
+
+                logger.info(f"MCP工具 {tool_name} 执行成功")
+                return {"success": True, "content": result_text, "raw_result": result}
+
+        except aiohttp.ClientError as e:
+            logger.error(f"调用MCP工具失败（网络错误）: {e}")
+            return {"success": False, "error": str(e), "content": f"网络错误：无法调用工具 {tool_name}"}
+        except Exception as e:
+            logger.error(f"调用MCP工具时发生错误: {e}")
+            return {"success": False, "error": str(e), "content": f"调用工具 {tool_name} 时发生错误"}
+
+    async def list_resources(self) -> list[dict[str, Any]]:
+        """
+        获取MCP服务器提供的资源列表
+
+        Returns:
+            List[Dict]: 资源列表
+        """
+        logger.info(f"正在从MCP服务器获取资源列表: {self.server_url}")
+
+        try:
+            session = await self._get_session()
+            url = f"{self.server_url}/resources/list"
+
+            async with session.post(url, headers=self._build_headers(), json={}) as response:
+                if response.status != 200:
+                    error_text = await response.text()
+                    logger.error(f"获取MCP资源列表失败: HTTP {response.status} - {error_text}")
+                    return []
+
+                data = await response.json()
+                resources = data.get("resources", [])
+
+                logger.info(f"成功获取 {len(resources)} 个MCP资源")
+                return resources
+
+        except Exception as e:
+            logger.error(f"获取MCP资源列表时发生错误: {e}")
+            return []
+
+    async def read_resource(self, resource_uri: str) -> dict[str, Any]:
+        """
+        读取MCP资源
+
+        Args:
+            resource_uri: 资源URI
+
+        Returns:
+            Dict: 资源内容
+        """
+        logger.info(f"读取MCP资源: {resource_uri}")
+
+        try:
+            session = await self._get_session()
+            url = f"{self.server_url}/resources/read"
+
+            payload = {"uri": resource_uri}
+
+            async with session.post(url, headers=self._build_headers(), json=payload) as response:
+                if response.status != 200:
+                    error_text = await response.text()
+                    logger.error(f"读取MCP资源失败: HTTP {response.status} - {error_text}")
+                    return {"success": False, "error": error_text}
+
+                result = await response.json()
+                logger.info(f"成功读取MCP资源: {resource_uri}")
+                return {"success": True, "content": result.get("contents", [])}
+
+        except Exception as e:
+            logger.error(f"读取MCP资源时发生错误: {e}")
+            return {"success": False, "error": str(e)}

src/plugin_system/utils/mcp_tool_provider.py

@@ -0,0 +1,174 @@
+"""
+MCP工具提供器 - 简化版
+直接集成到工具系统，无需复杂的插件架构
+"""
+
+import asyncio
+from typing import Any
+
+from src.common.logger import get_logger
+from src.plugin_system.utils.mcp_connector import MCPConnector
+
+logger = get_logger("MCP工具提供器")
+
+
+class MCPToolProvider:
+    """MCP工具提供器单例"""
+
+    _instance = None
+    _initialized = False
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
+    def __init__(self):
+        if not MCPToolProvider._initialized:
+            self.connectors: dict[str, MCPConnector] = {}
+            self.mcp_tools: dict[str, dict[str, Any]] = {}
+            """格式: {tool_full_name: {"connector": connector, "original_name": name, "definition": def}}"""
+            MCPToolProvider._initialized = True
+
+    async def initialize(self, mcp_servers: list[dict]):
+        """
+        初始化MCP服务器连接
+
+        Args:
+            mcp_servers: MCP服务器配置列表
+        """
+        logger.info(f"初始化MCP工具提供器，共{len(mcp_servers)}个服务器")
+
+        for server_config in mcp_servers:
+            await self._connect_server(server_config)
+
+        logger.info(f"MCP工具提供器初始化完成，共注册{len(self.mcp_tools)}个工具")
+
+    async def _connect_server(self, config: dict):
+        """连接单个MCP服务器"""
+        name = config.get("name", "unnamed")
+        url = config.get("url")
+        api_key = config.get("api_key")
+        enabled = config.get("enabled", True)
+
+        if not enabled or not url:
+            return
+
+        logger.info(f"连接MCP服务器: {name} ({url})")
+
+        connector = MCPConnector(url, api_key, config.get("timeout", 30))
+        self.connectors[name] = connector
+
+        try:
+            tools = await connector.list_tools()
+
+            for tool_name, tool_def in tools.items():
+                # 使用服务器名作前缀
+                full_name = f"{name}_{tool_name}"
+                self.mcp_tools[full_name] = {
+                    "connector": connector,
+                    "original_name": tool_name,
+                    "definition": tool_def,
+                    "server_name": name,
+                }
+
+            logger.info(f"从{name}获取{len(tools)}个工具")
+
+        except Exception as e:
+            logger.error(f"连接MCP服务器{name}失败: {e}")
+
+    def get_mcp_tool_definitions(self) -> list[tuple[str, dict[str, Any]]]:
+        """
+        获取所有MCP工具定义（适配Bot的工具格式）
+
+        Returns:
+            List[Tuple[str, dict]]: [(tool_name, tool_definition), ...]
+        """
+        definitions = []
+
+        for full_name, tool_info in self.mcp_tools.items():
+            mcp_def = tool_info["definition"]
+            input_schema = mcp_def.get("input_schema", {})
+
+            # 转换为Bot的工具格式
+            bot_tool_def = {
+                "name": full_name,
+                "description": mcp_def.get("description", f"MCP工具: {full_name}"),
+                "parameters": self._convert_schema_to_parameters(input_schema),
+            }
+
+            definitions.append((full_name, bot_tool_def))
+
+        return definitions
+
+    def _convert_schema_to_parameters(self, schema: dict) -> list[tuple]:
+        """
+        将MCP的JSON Schema转换为Bot的参数格式
+
+        Args:
+            schema: MCP的inputSchema
+
+        Returns:
+            Bot的parameters格式
+        """
+        from src.plugin_system.base.component_types import ToolParamType
+
+        parameters = []
+        properties = schema.get("properties", {})
+        required = schema.get("required", [])
+
+        type_mapping = {
+            "string": ToolParamType.STRING,
+            "integer": ToolParamType.INTEGER,
+            "number": ToolParamType.FLOAT,
+            "boolean": ToolParamType.BOOLEAN,
+        }
+
+        for param_name, param_def in properties.items():
+            param_type = type_mapping.get(param_def.get("type", "string"), ToolParamType.STRING)
+            description = param_def.get("description", "")
+            is_required = param_name in required
+            enum_values = param_def.get("enum", None)
+
+            parameters.append((param_name, param_type, description, is_required, enum_values))
+
+        return parameters
+
+    async def call_mcp_tool(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+        """
+        调用MCP工具
+
+        Args:
+            tool_name: 工具全名（包含前缀）
+            arguments: 参数
+
+        Returns:
+            工具执行结果
+        """
+        if tool_name not in self.mcp_tools:
+            return {"content": f"MCP工具{tool_name}不存在"}
+
+        tool_info = self.mcp_tools[tool_name]
+        connector = tool_info["connector"]
+        original_name = tool_info["original_name"]
+
+        logger.info(f"调用MCP工具: {tool_name}")
+
+        result = await connector.call_tool(original_name, arguments)
+
+        if result.get("success"):
+            return {"content": result.get("content", "")}
+        else:
+            return {"content": f"工具执行失败: {result.get('error', '未知错误')}"}
+
+    async def close(self):
+        """关闭所有连接"""
+        for name, connector in self.connectors.items():
+            try:
+                await connector.close()
+            except Exception as e:
+                logger.error(f"关闭MCP连接{name}失败: {e}")
+
+
+# 全局单例
+mcp_tool_provider = MCPToolProvider()

src/plugins/built_in/affinity_flow_chatter/plan_executor.py

@@ -441,7 +441,7 @@ class ChatterPlanExecutor:
             # 通过 chat_id 获取真实的 chat_stream 对象
             from src.plugin_system.apis.chat_api import get_chat_manager
             chat_manager = get_chat_manager()
-            chat_stream = chat_manager.get_stream(plan.chat_id)
+            chat_stream = await chat_manager.get_stream(plan.chat_id)
 
             if chat_stream:
                 # 调用 action_manager 的批量存储

src/plugins/built_in/proactive_thinker/proacive_thinker_event.py

@@ -11,7 +11,7 @@ from src.common.logger import get_logger
 from src.config.config import global_config
 from src.manager.async_task_manager import AsyncTask, async_task_manager
 from src.plugin_system import BaseEventHandler, EventType
-from src.plugin_system.apis import chat_api, person_api
+from src.plugin_system.apis import chat_api, message_api, person_api
 from src.plugin_system.base.base_event import HandlerResult
 
 from .proactive_thinker_executor import ProactiveThinkerExecutor
@@ -160,7 +160,9 @@ class ProactiveThinkingTask(AsyncTask):
                             continue
 
                         # 检查冷却时间
-                        time_since_last_active = time.time() - stream.last_active_time
+                        recent_messages = await message_api.get_recent_messages(chat_id=stream.stream_id, limit=1,limit_mode="latest")
+                        last_message_time = recent_messages[0]["time"] if recent_messages else stream.create_time
+                        time_since_last_active = time.time() - last_message_time
                         if time_since_last_active > next_interval:
                             logger.info(
                                 f"【日常唤醒-私聊】聊天流 {stream.stream_id} 已冷却 {time_since_last_active:.2f} 秒，触发主动对话。"
@@ -184,7 +186,9 @@ class ProactiveThinkingTask(AsyncTask):
                     # 检查群聊是否在白名单内
                     if not enabled_groups or f"qq:{stream.group_info.group_id}" in enabled_groups:
                         # 检查冷却时间
-                        time_since_last_active = time.time() - stream.last_active_time
+                        recent_messages = await message_api.get_recent_messages(chat_id=stream.stream_id, limit=1)
+                        last_message_time = recent_messages[0]["time"] if recent_messages else stream.create_time
+                        time_since_last_active = time.time() - last_message_time
                         if time_since_last_active > next_interval:
                             logger.info(
                                 f"【日常唤醒-群聊】聊天流 {stream.stream_id} 已冷却 {time_since_last_active:.2f} 秒，触发主动对话。"

src/plugins/built_in/proactive_thinker/proactive_thinker_executor.py

@@ -1,8 +1,10 @@
+import time
 from datetime import datetime
 from typing import Any
 
 import orjson
 
+from src.chat.utils.chat_message_builder import build_readable_actions, get_actions_by_timestamp_with_chat
 from src.common.logger import get_logger
 from src.config.config import global_config, model_config
 from src.mood.mood_manager import mood_manager
@@ -140,23 +142,20 @@ class ProactiveThinkerExecutor:
             else "今天没有日程安排。"
         )
 
-        recent_messages = await message_api.get_recent_messages(stream.stream_id)
+        recent_messages = await message_api.get_recent_messages(stream.stream_id,limit=50,limit_mode="latest",hours=12)
         recent_chat_history = (
             await message_api.build_readable_messages_to_str(recent_messages) if recent_messages else "无"
         )
 
-        action_history_list = await database_api.db_query(
-            database_api.MODEL_MAPPING["ActionRecords"],
-            filters={"chat_id": stream_id, "action_name": "proactive_decision"},
-            limit=3,
-            order_by=["-time"],
-        )
-        action_history_context = (
-            "\n".join([f"- {a['action_data']}" for a in action_history_list if isinstance(a, dict)])
-            if isinstance(action_history_list, list)
-            else "无"
+        action_history_list = await get_actions_by_timestamp_with_chat(
+            chat_id=stream.stream_id,
+            timestamp_start=time.time() - 3600 * 24,  #过去24小时
+            timestamp_end=time.time(),
+            limit=7,
         )
 
+        action_history_context = build_readable_actions(actions=action_history_list)
+
         # 2. 构建基础上下文
         mood_state = "暂时没有"
         if global_config.mood.enable_mood: