get_tools_from_mcp

agent/deep_assistant.py

@@ -21,22 +21,38 @@ def read_mcp_settings():
         mcp_settings_json = json.load(f)
     return mcp_settings_json
 
+async def get_tools_from_mcp(mcp):
+    """从MCP配置中提取工具"""
+    # 防御式处理：确保 mcp 是列表且长度大于 0，且包含 mcpServers
+    if not isinstance(mcp, list) or len(mcp) == 0 or "mcpServers" not in mcp[0]:
+        return []
+
+    # 修改 mcp[0]["mcpServers"] 列表，把 type 字段改成 transport
+    # 如果没有 transport，则根据是否存在 url 默认 transport 为 http 或 stdio
+    for cfg in mcp[0]["mcpServers"].values():
+        if "type" in cfg:
+            cfg.pop("type")
+        if "transport" not in cfg:
+            cfg["transport"] = "http" if "url" in cfg else "stdio"
+
+    # 确保 mcp[0]["mcpServers"] 是字典类型
+    if not isinstance(mcp[0]["mcpServers"], dict):
+        return []
+
+    try:
+        mcp_client = MultiServerMCPClient(mcp[0]["mcpServers"])
+        mcp_tools = await mcp_client.get_tools()
+        return mcp_tools
+    except Exception:
+        # 发生异常时返回空列表，避免上层调用报错
+        return []
 
 async def init_agent(bot_id: str, model_name="qwen3-next", api_key=None,
                     model_server=None, generate_cfg=None,
                     system_prompt=None, mcp=None, robot_type=None, language="jp", user_identifier=None):
     system = system_prompt if system_prompt else read_system_prompt()
     mcp = mcp if mcp else read_mcp_settings()
-    # 修改mcp[0]["mcpServers"]列表，把 type 字段改成 transport ，如果没有的话，就默认transport:stdio
+    mcp_tools = await get_tools_from_mcp(mcp)
-    if mcp and "mcpServers" in mcp[0]:
-        for cfg in mcp[0]["mcpServers"].values():
-            if "type" in cfg:
-                cfg.pop("type")
-            if "transport" not in cfg:
-                cfg["transport"] = "http" if "url" in cfg else "stdio"
-
-    mcp_client = MultiServerMCPClient(mcp[0]["mcpServers"])
-    mcp_tools = await mcp_client.get_tools()
 
     # 检测或使用指定的提供商
     model_provider,base_url = detect_provider(model_name,model_server)