mirror of
https://github.com/AstrBotDevs/AstrBot.git
synced 2026-10-24 23:21:40 +08:00
feat(mcp): enhance logging and initialize MCP clients in background (#5993)
* feat(mcp): enhance logging and initialize MCP clients in background fixes: #5777 * rf * fix(mcp): simplify MCP client initialization in background * fix(mcp): update error message for MCP background initialization failure
This commit is contained in:
@@ -144,10 +144,14 @@ class MCPClient:
|
||||
|
||||
cfg = _prepare_config(mcp_server_config.copy())
|
||||
|
||||
def logging_callback(msg: str) -> None:
|
||||
def logging_callback(
|
||||
msg: str | mcp.types.LoggingMessageNotificationParams,
|
||||
) -> None:
|
||||
# Handle MCP service error logs
|
||||
print(f"MCP Server {name} Error: {msg}")
|
||||
self.server_errlogs.append(msg)
|
||||
if isinstance(msg, mcp.types.LoggingMessageNotificationParams):
|
||||
if msg.level in ("warning", "error", "critical", "alert", "emergency"):
|
||||
log_msg = f"[{msg.level.upper()}] {str(msg.data)}"
|
||||
self.server_errlogs.append(log_msg)
|
||||
|
||||
if "url" in cfg:
|
||||
success, error_msg = await _quick_test_mcp_connection(cfg)
|
||||
@@ -214,15 +218,24 @@ class MCPClient:
|
||||
**cfg,
|
||||
)
|
||||
|
||||
def callback(msg: str) -> None:
|
||||
def callback(msg: str | mcp.types.LoggingMessageNotificationParams) -> None:
|
||||
# Handle MCP service error logs
|
||||
self.server_errlogs.append(msg)
|
||||
if isinstance(msg, mcp.types.LoggingMessageNotificationParams):
|
||||
if msg.level in (
|
||||
"warning",
|
||||
"error",
|
||||
"critical",
|
||||
"alert",
|
||||
"emergency",
|
||||
):
|
||||
log_msg = f"[{msg.level.upper()}] {str(msg.data)}"
|
||||
self.server_errlogs.append(log_msg)
|
||||
|
||||
stdio_transport = await self.exit_stack.enter_async_context(
|
||||
mcp.stdio_client(
|
||||
server_params,
|
||||
errlog=LogPipe(
|
||||
level=logging.ERROR,
|
||||
level=logging.INFO,
|
||||
logger=logger,
|
||||
identifier=f"MCPServer-{name}",
|
||||
callback=callback,
|
||||
|
||||
@@ -21,8 +21,8 @@ from astrbot.core.utils.astrbot_path import get_astrbot_data_path
|
||||
|
||||
DEFAULT_MCP_CONFIG = {"mcpServers": {}}
|
||||
|
||||
DEFAULT_MCP_INIT_TIMEOUT_SECONDS = 20.0
|
||||
DEFAULT_ENABLE_MCP_TIMEOUT_SECONDS = 30.0
|
||||
DEFAULT_MCP_INIT_TIMEOUT_SECONDS = 180.0
|
||||
DEFAULT_ENABLE_MCP_TIMEOUT_SECONDS = 180.0
|
||||
MCP_INIT_TIMEOUT_ENV = "ASTRBOT_MCP_INIT_TIMEOUT"
|
||||
ENABLE_MCP_TIMEOUT_ENV = "ASTRBOT_MCP_ENABLE_TIMEOUT"
|
||||
MAX_MCP_TIMEOUT_SECONDS = 300.0
|
||||
@@ -417,9 +417,11 @@ class FunctionToolManager:
|
||||
for (name, cfg, _), result in zip(active_configs, results, strict=False):
|
||||
if isinstance(result, Exception):
|
||||
if isinstance(result, MCPInitTimeoutError):
|
||||
logger.error(f"MCP 服务 {name} 初始化超时({timeout_display}秒)")
|
||||
logger.error(
|
||||
f"Connected to MCP server {name} timeout ({timeout_display} seconds)"
|
||||
)
|
||||
else:
|
||||
logger.error(f"MCP 服务 {name} 初始化失败: {result}")
|
||||
logger.error(f"Failed to initialize MCP server {name}: {result}")
|
||||
self._log_safe_mcp_debug_config(cfg)
|
||||
failed_services.append(name)
|
||||
async with self._runtime_lock:
|
||||
@@ -430,16 +432,18 @@ class FunctionToolManager:
|
||||
|
||||
if failed_services:
|
||||
logger.warning(
|
||||
f"以下 MCP 服务初始化失败: {', '.join(failed_services)}。"
|
||||
f"请检查配置文件 mcp_server.json 和服务器可用性。"
|
||||
f"The following MCP services failed to initialize: {', '.join(failed_services)}. "
|
||||
f"Please check the mcp_server.json file and server availability."
|
||||
)
|
||||
|
||||
summary = MCPInitSummary(
|
||||
total=len(active_configs), success=success_count, failed=failed_services
|
||||
)
|
||||
logger.info(f"MCP 服务初始化完成: {summary.success}/{summary.total} 成功")
|
||||
logger.info(
|
||||
f"MCP services initialization completed: {summary.success}/{summary.total} successful, {len(summary.failed)} failed."
|
||||
)
|
||||
if summary.total > 0 and summary.success == 0:
|
||||
msg = "全部 MCP 服务初始化失败,请检查 mcp_server.json 配置和服务器可用性。"
|
||||
msg = "All MCP services failed to initialize, please check the mcp_server.json and server availability."
|
||||
if raise_on_all_failed:
|
||||
raise MCPAllServicesFailedError(msg)
|
||||
logger.error(msg)
|
||||
@@ -461,7 +465,7 @@ class FunctionToolManager:
|
||||
async with self._runtime_lock:
|
||||
if name in self._mcp_server_runtime or name in self._mcp_starting:
|
||||
logger.warning(
|
||||
f"MCP 服务 {name} 已在运行,忽略本次启用请求(timeout={timeout:g})。"
|
||||
f"Connected to MCP server {name}, ignoring this startup request (timeout={timeout:g})."
|
||||
)
|
||||
self._log_safe_mcp_debug_config(cfg)
|
||||
return
|
||||
@@ -478,10 +482,10 @@ class FunctionToolManager:
|
||||
)
|
||||
except asyncio.TimeoutError as exc:
|
||||
raise MCPInitTimeoutError(
|
||||
f"MCP 服务 {name} 初始化超时({timeout:g} 秒)"
|
||||
f"Connected to MCP server {name} timeout ({timeout:g} seconds)"
|
||||
) from exc
|
||||
except Exception:
|
||||
logger.error(f"初始化 MCP 客户端 {name} 失败", exc_info=True)
|
||||
logger.error(f"Failed to initialize MCP client {name}", exc_info=True)
|
||||
raise
|
||||
finally:
|
||||
if mcp_client is None:
|
||||
@@ -491,9 +495,9 @@ class FunctionToolManager:
|
||||
async def lifecycle() -> None:
|
||||
try:
|
||||
await shutdown_event.wait()
|
||||
logger.info(f"收到 MCP 客户端 {name} 终止信号")
|
||||
logger.info(f"Received shutdown signal for MCP client {name}")
|
||||
except asyncio.CancelledError:
|
||||
logger.debug(f"MCP 客户端 {name} 任务被取消")
|
||||
logger.debug(f"MCP client {name} task was cancelled")
|
||||
raise
|
||||
finally:
|
||||
await self._terminate_mcp_client(name)
|
||||
@@ -545,7 +549,7 @@ class FunctionToolManager:
|
||||
if strict:
|
||||
raise MCPShutdownTimeoutError(pending_names, timeout)
|
||||
logger.warning(
|
||||
"MCP 服务关闭超时(%s 秒),以下服务未完全关闭:%s",
|
||||
"MCP server shutdown timeout (%s seconds), the following servers were not fully closed: %s",
|
||||
f"{timeout:g}",
|
||||
", ".join(pending_names),
|
||||
)
|
||||
@@ -568,7 +572,9 @@ class FunctionToolManager:
|
||||
try:
|
||||
await mcp_client.cleanup()
|
||||
except Exception as cleanup_exc: # noqa: BLE001 - only log here
|
||||
logger.error(f"清理 MCP 客户端资源 {name} 失败: {cleanup_exc}")
|
||||
logger.error(
|
||||
f"Failed to cleanup MCP client resources {name}: {cleanup_exc}"
|
||||
)
|
||||
|
||||
async def _init_mcp_client(self, name: str, config: dict) -> MCPClient:
|
||||
"""初始化单个MCP客户端"""
|
||||
@@ -602,7 +608,7 @@ class FunctionToolManager:
|
||||
)
|
||||
self.func_list.append(func_tool)
|
||||
|
||||
logger.info(f"已连接 MCP 服务 {name}, Tools: {tool_names}")
|
||||
logger.info(f"Connected to MCP server {name}, Tools: {tool_names}")
|
||||
return mcp_client
|
||||
|
||||
async def _terminate_mcp_client(self, name: str) -> None:
|
||||
@@ -622,7 +628,7 @@ class FunctionToolManager:
|
||||
async with self._runtime_lock:
|
||||
self._mcp_server_runtime.pop(name, None)
|
||||
self._mcp_starting.discard(name)
|
||||
logger.info(f"已关闭 MCP 服务 {name}")
|
||||
logger.info(f"Disconnected from MCP server {name}")
|
||||
return
|
||||
|
||||
# Runtime missing but stale tools may still exist after failed flows.
|
||||
|
||||
@@ -79,6 +79,7 @@ class ProviderManager:
|
||||
self._provider_change_hooks: list[
|
||||
Callable[[str, ProviderType, str | None], None]
|
||||
] = []
|
||||
self._mcp_init_task: asyncio.Task | None = None
|
||||
|
||||
def set_provider_change_callback(
|
||||
self,
|
||||
@@ -330,24 +331,16 @@ class ProviderManager:
|
||||
if not self.curr_tts_provider_inst and self.tts_provider_insts:
|
||||
self.curr_tts_provider_inst = self.tts_provider_insts[0]
|
||||
|
||||
# 初始化 MCP Client 连接(等待完成以确保工具可用)
|
||||
strict_mcp_init = os.getenv("ASTRBOT_MCP_INIT_STRICT", "").strip().lower() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
"on",
|
||||
}
|
||||
mcp_init_summary = await self.llm_tools.init_mcp_clients(
|
||||
raise_on_all_failed=strict_mcp_init
|
||||
)
|
||||
if (
|
||||
mcp_init_summary.total > 0
|
||||
and mcp_init_summary.success == 0
|
||||
and not strict_mcp_init
|
||||
):
|
||||
logger.warning(
|
||||
"MCP 服务全部初始化失败,系统将继续启动(可设置 "
|
||||
"ASTRBOT_MCP_INIT_STRICT=1 以在此场景下中止启动)。"
|
||||
async def _init_mcp_clients_bg() -> None:
|
||||
try:
|
||||
await self.llm_tools.init_mcp_clients()
|
||||
except Exception:
|
||||
logger.error("MCP init background task failed", exc_info=True)
|
||||
|
||||
if self._mcp_init_task is None or self._mcp_init_task.done():
|
||||
self._mcp_init_task = asyncio.create_task(
|
||||
_init_mcp_clients_bg(),
|
||||
name="provider-manager:mcp-init",
|
||||
)
|
||||
|
||||
def dynamic_import_provider(self, type: str) -> None:
|
||||
@@ -817,6 +810,13 @@ class ProviderManager:
|
||||
await self.load_provider(new_config)
|
||||
|
||||
async def terminate(self) -> None:
|
||||
if self._mcp_init_task and not self._mcp_init_task.done():
|
||||
self._mcp_init_task.cancel()
|
||||
try:
|
||||
await self._mcp_init_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
for provider_inst in self.provider_insts:
|
||||
if hasattr(provider_inst, "terminate"):
|
||||
await provider_inst.terminate() # type: ignore
|
||||
|
||||
Reference in New Issue
Block a user