Skip to content
Merged
91 changes: 85 additions & 6 deletions astrbot/core/agent/mcp_client.py
Comment thread
Soulter marked this conversation as resolved.
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@
import logging
import os
import sys
import subprocess
import shutil
from contextlib import AsyncExitStack
from datetime import timedelta
from typing import Generic
Expand Down Expand Up @@ -51,15 +53,92 @@ def _prepare_stdio_env(config: dict) -> dict:
"""Preserve Windows executable resolution for stdio subprocesses."""
if sys.platform != "win32":
return config

pathext = os.environ.get("PATHEXT")
if not pathext:
return config

prepared = config.copy()
env = dict(prepared.get("env") or {})
env.setdefault("PATHEXT", pathext)
env= _merge_environment_variables(env)
prepared["env"] = env
# 获取配置值,并转换为小写进行不区分大小写比较
cmd_str = _extract_command_string(config)
# 目前仅处理 dotnet,如有其他命令需求需扩展
if cmd_str and cmd_str.lower() == "dotnet":
env= _ensure_dotnet_in_path(env)
prepared["env"] = env
return _create_subprocess_NO_WINDOW(prepared)
return prepared

def _extract_command_string(config:dict)->str:
"""从配置中提取命令字符串"""
command = config.get('command')
cmd_str = ""
if isinstance(command, str):
cmd_str = command
elif isinstance(command, list) and command:
cmd_str = command[0]
return cmd_str

def _merge_environment_variables(env: dict) -> dict:
"""合并环境变量,处理Windows不区分大小写的情况"""
merged = env.copy()

# 将用户环境变量转换为统一的大小写形式便于比较
user_keys_lower = {k.lower(): k for k in merged.keys()}

for sys_key, sys_value in os.environ.items():
sys_key_lower = sys_key.lower()
if sys_key_lower not in user_keys_lower:
# 使用系统环境变量中的原始大小写
merged[sys_key] = sys_value

return merged

def _ensure_dotnet_in_path(env: dict) -> dict:
"""确保dotnet在PATH中,若不存在则发出警告而不是自动添加"""

# 检查当前环境PATH中是否有dotnet
current_path = env.get("PATH", "")
if shutil.which("dotnet", path=current_path):
return env

# 检查系统PATH
system_path = os.environ.get("PATH", "")
if shutil.which("dotnet", path=system_path):
# 安全地合并PATH:过滤空值后连接
paths = [p for p in [current_path, system_path] if p]
env["PATH"] = ";".join(paths)
return env

# 发出警告而不是静默添加
logger.warning(
"dotnet not found in PATH. .NET-based MCP servers may fail to start. "
"Please ensure dotnet is installed and in your PATH."
)
return env

def _create_subprocess(prepared: dict)->dict:
"""准备子进程参数字典"""
# 只在Windows平台处理
if os.name == "nt":
# 检查用户是否指定了控制台行为
# 使用更清晰的参数名:show_console
show_console = prepared.pop("show_console", None)

if show_console is not None:
# 确保是布尔值
if isinstance(show_console, str):
# 处理字符串形式的布尔值
show_console = show_console.lower() in ("true", "1", "yes", "y", "t")
elif not isinstance(show_console, bool):
# 如果不是字符串也不是布尔值,可以抛出异常或使用默认值
raise ValueError(f"show_console must be bool or str, got {type(show_console)}")
existing_flags = prepared.get("creationflags", 0)
if not show_console: # 如果不要显示控制台
prepared["creationflags"] = existing_flags | subprocess.CREATE_NO_WINDOW
else:
prepared["creationflags"] = existing_flags & ~subprocess.CREATE_NO_WINDOW
else:
# 保持向后兼容:默认添加CREATE_NO_WINDOW
if "creationflags" not in prepared:
prepared["creationflags"] = subprocess.CREATE_NO_WINDOW
return prepared


Expand Down