mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(mcp): run lazy MCP discovery once when initialization itself raises (#5836)
* fix(mcp): run lazy MCP discovery once when initialization itself raises get_cached_mcp_tools() wrapped the whole lazy-initialization call in the `except RuntimeError` that was only meant to catch asyncio.get_event_loop() raising when no loop is set. run_until_complete() and Future.result() re-raise the coroutine's own exception, so a RuntimeError from discovery — e.g. the deliberately re-raised McpTaskConfigurationError — took the same fallback and ran asyncio.run(initialize_mcp_tools()) a second time, respawning every stdio server and re-fetching OAuth tokens before giving up. Inside a running loop that fallback fails with "asyncio.run() cannot be called from a running event loop", so the logged traceback hid the real cause. Resolve the loop first, in its own try/except, and drive initialization exactly once: asyncio.run when no loop is set or the current loop is closed, a worker thread when the loop is running, run_until_complete otherwise. Any failure is logged once and returns [] as before. Regression tests pin one discovery call and the real logged cause with and without a running loop, and keep both fallbacks (no loop set, closed loop) covered. * test(mcp): surface worker-thread failures in the no-loop fallback test Thread.join() does not propagate exceptions, so a regression of the asyncio.run fallback failed this test with a bare KeyError: 'result' instead of the real traceback. Capture sys.exc_info() in the worker and re-raise it on the main thread before the assertions.
This commit is contained in:
parent
c943d953cd
commit
174c6f7fd3
@ -965,6 +965,13 @@ This release closes that milestone with **765 merged pull requests**.
|
||||
the API had reported as paused. The read now takes the writer first, as every
|
||||
other mutating path in that repository does. PostgreSQL was unaffected.
|
||||
([#5777])
|
||||
- **mcp:** Lazy MCP initialization no longer runs tool discovery twice when
|
||||
discovery itself raises a `RuntimeError` such as `McpTaskConfigurationError`.
|
||||
The `asyncio.run` fallback in `get_cached_mcp_tools()` was meant only for
|
||||
`get_event_loop()` failing, but it also caught discovery errors and
|
||||
re-spawned every stdio server (and re-fetched OAuth tokens) before giving
|
||||
up; inside a running loop it also logged a misleading "asyncio.run() cannot
|
||||
be called from a running event loop" traceback instead of the real cause.
|
||||
- **uploads:** Deleting an uploaded document no longer deletes the converted
|
||||
Markdown beside it. Conversion names a companion after the document's stem
|
||||
and falls back to a `_N` suffix when that name is taken, so the `.md` next to
|
||||
|
||||
@ -732,6 +732,12 @@
|
||||
SQLite 的写锁,因此过期的读取会通过校验,并把任务状态写回 `enabled` 且不改动 `next_run_at`,
|
||||
导致接口已回复“已暂停”的任务仍被继续触发。现在该读取会像该仓储中其他写入路径一样先获取写锁。
|
||||
PostgreSQL 不受影响。([#5777])
|
||||
- **mcp:** MCP 延迟初始化在工具发现本身抛出 `RuntimeError`(例如
|
||||
`McpTaskConfigurationError`)时,不再把发现流程跑两遍。`get_cached_mcp_tools()` 里的
|
||||
`asyncio.run` 兜底只为 `get_event_loop()` 失败而设,却同时捕获了发现阶段的错误,于是在
|
||||
放弃之前会重新拉起每一个 stdio 服务器(并重新获取 OAuth 令牌);在运行中的事件循环里,
|
||||
它记录的还是误导性的 "asyncio.run() cannot be called from a running event loop"
|
||||
堆栈,而不是真正的原因。
|
||||
- **上传:** 删除已上传的文档时,不再连带删除其旁边转换生成的 Markdown。转换以文档主干名
|
||||
命名配套文件,名称被占用时回退为 `_N` 后缀,因此文档旁的 `.md` 可能属于主干名相同的另一个
|
||||
文档,或属于用户自己:上传 `a.docx` 与 `a.pdf` 会生成 `a.md` 与 `a_1.md`,删除 `a.pdf`
|
||||
|
||||
@ -6,7 +6,10 @@
|
||||
- **Long-running ordinary task driver**: `extensions_config.json -> mcpServers.<server>.task_toolsets` binds exact raw submit/status/cancel names; one raw tool may occupy only one role across that server's groups. `mcp/tools.py` hides status/cancel and replaces submit with a wrapper that returns only the local task ID after persistence. `ordinary.py` reads only MCP `structuredContent`, maps remote `running` to `working`, and treats `error_code=task_not_found` or malformed structured output as permanent failure. A status call with `isError=true` is a retryable call failure: the first text content block is retained as a bounded diagnostic, while a permanent remote-task outcome must arrive in a normal result with structured `status=failed`. `task_tool_caller.py` restores the same canonical `(server_name, versioned user/thread/incarnation scope)` stdio session as ordinary calls; non-NULL incarnations use an unambiguous versioned encoding, while an explicitly captured legacy NULL retains the pre-activation `user_id:thread_id` scope. A runtime with a missing/invalid incarnation key fails closed. HTTP/SSE calls remain ephemeral, apply `session_init_timeout` to initialization and `tool_call_timeout` to task calls, and support server-level OAuth refresh outside an Agent run. `McpTaskService` exponentially backs off transient status/cancel errors without a maximum attempt count, derives API `tracking_degraded` from the consecutive-error threshold, keeps `input_required` on a slower poll, and caps finite positive remote poll hints at 24 hours. Task-enabled server runtime/binding configuration and `mcpInterceptors` are frozen to the Gateway startup snapshot; hot drift fails clearly before tool discovery can diverge from background calls, while presentation-only fields and non-task servers remain reloadable. Configured task toolsets fail startup when the runtime is disabled or persistence is memory. Users still cannot submit an answer back to an `input_required` remote task.
|
||||
- **Durable task payload bounds**: persisted task errors are capped at 4,000 characters. `input_required` and `result_artifact` must each serialize as valid JSON within 64 KiB; an invalid or oversized payload becomes a permanent protocol failure rather than being truncated and changing its semantics. Remote task IDs/task names are limited to 255 characters and task-enabled server names to 128, matching the SQL schema; an oversized submitted remote ID is rejected only after the Service has the handle so compensation cancellation still runs. Oversized results retain the existing bounded preview/truncation/artifact behavior.
|
||||
- **Remote task bring-up**: One `session_init_timeout` deadline covers transport context entry and MCP initialization. Report the server and configured bound on expiry; disable it before calling the tool. Enter and exit the adapter's AnyIO context managers in the same task; tests include real SSE endpoint/initialization stalls and connection cleanup.
|
||||
- **Lazy initialization**: Tools loaded on first use via `get_cached_mcp_tools()`
|
||||
- **Lazy initialization**: Tools loaded on first use via `get_cached_mcp_tools()`. Its
|
||||
`asyncio.run` fallback covers only `get_event_loop()` raising (no loop set, or a closed
|
||||
loop); a `RuntimeError` from discovery itself (e.g. `McpTaskConfigurationError`) is
|
||||
logged once and never triggers a second discovery pass.
|
||||
- **Loop-isolated stdio sessions**: Both the live registry and in-flight creations
|
||||
are keyed by `(server_name, scope_key, owning_loop)`. Same-loop callers share
|
||||
initialization and state; live sibling loops must never cancel or replace each
|
||||
|
||||
@ -381,9 +381,18 @@ def get_cached_mcp_tools() -> list[BaseTool]:
|
||||
retired_pool.close_all_sync()
|
||||
|
||||
logger.info("MCP tools not initialized, performing lazy initialization...")
|
||||
# Only ``get_event_loop()`` may fall back to ``asyncio.run``: a
|
||||
# ``RuntimeError`` raised *by* ``initialize_mcp_tools()`` (for example
|
||||
# ``McpTaskConfigurationError``) must not trigger a second discovery
|
||||
# pass that respawns every stdio server and re-fetches OAuth tokens.
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
if loop.is_running():
|
||||
loop: asyncio.AbstractEventLoop | None = asyncio.get_event_loop()
|
||||
except RuntimeError:
|
||||
loop = None
|
||||
try:
|
||||
if loop is None or loop.is_closed():
|
||||
asyncio.run(initialize_mcp_tools())
|
||||
elif loop.is_running():
|
||||
import concurrent.futures
|
||||
|
||||
with concurrent.futures.ThreadPoolExecutor() as executor:
|
||||
@ -391,12 +400,6 @@ def get_cached_mcp_tools() -> list[BaseTool]:
|
||||
future.result()
|
||||
else:
|
||||
loop.run_until_complete(initialize_mcp_tools())
|
||||
except RuntimeError:
|
||||
try:
|
||||
asyncio.run(initialize_mcp_tools())
|
||||
except Exception:
|
||||
logger.exception("Failed to lazy-initialize MCP tools")
|
||||
return []
|
||||
except Exception:
|
||||
logger.exception("Failed to lazy-initialize MCP tools")
|
||||
return []
|
||||
|
||||
@ -24,6 +24,7 @@ import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
@ -1206,3 +1207,117 @@ def test_unverifiable_signature_is_not_treated_as_stable(cache_globals, monkeypa
|
||||
|
||||
assert cache_module._read_stable_mcp_snapshot(cfg, incomplete) is None
|
||||
assert cache_module._is_cache_stale() is True
|
||||
|
||||
|
||||
class TestLazyInitializationFailure:
|
||||
"""``get_cached_mcp_tools()`` must run discovery exactly once per call.
|
||||
|
||||
The ``except RuntimeError`` fallback exists for ``asyncio.get_event_loop()``
|
||||
raising when no loop is set (non-main thread, or Python 3.14+). It must not
|
||||
also catch a ``RuntimeError`` raised *by* ``initialize_mcp_tools()`` itself —
|
||||
``McpTaskConfigurationError`` is one, and it is deliberately re-raised by
|
||||
``get_mcp_tools`` — because the fallback then re-runs a full discovery pass
|
||||
(respawning every stdio server, re-fetching OAuth tokens) before giving up.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _install_failing_discovery(monkeypatch, tmp_path: Path) -> list[str]:
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(cfg, {"srv1": _server()})
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg))
|
||||
calls: list[str] = []
|
||||
|
||||
async def _fake_tools(**_kwargs):
|
||||
calls.append(threading.current_thread().name)
|
||||
raise RuntimeError("discovery failed")
|
||||
|
||||
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
|
||||
return calls
|
||||
|
||||
def test_runtime_error_from_discovery_without_running_loop_runs_once(self, cache_globals, monkeypatch, tmp_path, caplog):
|
||||
calls = self._install_failing_discovery(monkeypatch, tmp_path)
|
||||
|
||||
with caplog.at_level("ERROR", logger="deerflow.mcp.cache"):
|
||||
assert cache_module.get_cached_mcp_tools() == []
|
||||
|
||||
assert calls == [threading.current_thread().name]
|
||||
assert cache_module._cache_initialized is False
|
||||
logged = [record for record in caplog.records if record.exc_info]
|
||||
assert len(logged) == 1
|
||||
assert str(logged[0].exc_info[1]) == "discovery failed"
|
||||
|
||||
def test_runtime_error_from_discovery_inside_running_loop_runs_once(self, cache_globals, monkeypatch, tmp_path, caplog):
|
||||
calls = self._install_failing_discovery(monkeypatch, tmp_path)
|
||||
|
||||
async def _call_from_running_loop():
|
||||
return cache_module.get_cached_mcp_tools()
|
||||
|
||||
with caplog.at_level("ERROR", logger="deerflow.mcp.cache"):
|
||||
assert asyncio.run(_call_from_running_loop()) == []
|
||||
|
||||
assert len(calls) == 1
|
||||
assert cache_module._cache_initialized is False
|
||||
# The logged failure must be the discovery error, not a misleading
|
||||
# "asyncio.run() cannot be called from a running event loop".
|
||||
logged = [record for record in caplog.records if record.exc_info]
|
||||
assert len(logged) == 1
|
||||
assert str(logged[0].exc_info[1]) == "discovery failed"
|
||||
|
||||
def test_thread_without_event_loop_still_initializes_once(self, cache_globals, monkeypatch, tmp_path):
|
||||
"""The no-loop fallback stays intact: a worker thread with no loop set initializes via ``asyncio.run``."""
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(cfg, {"srv1": _server()})
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg))
|
||||
calls: list[str] = []
|
||||
|
||||
async def _fake_tools(**_kwargs):
|
||||
calls.append(threading.current_thread().name)
|
||||
return []
|
||||
|
||||
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
|
||||
outcome: dict[str, object] = {}
|
||||
|
||||
def _worker():
|
||||
# ``Thread.join`` does not propagate exceptions: capture them so a
|
||||
# regression surfaces its real traceback instead of a bare KeyError.
|
||||
try:
|
||||
with pytest.raises(RuntimeError):
|
||||
asyncio.get_event_loop()
|
||||
outcome["result"] = cache_module.get_cached_mcp_tools()
|
||||
except BaseException:
|
||||
outcome["error"] = sys.exc_info()
|
||||
|
||||
worker = threading.Thread(target=_worker, name="mcp-no-loop-worker")
|
||||
worker.start()
|
||||
worker.join(timeout=10)
|
||||
assert not worker.is_alive()
|
||||
if "error" in outcome:
|
||||
_, exc, tb = outcome["error"]
|
||||
raise exc.with_traceback(tb)
|
||||
|
||||
assert outcome["result"] == []
|
||||
assert len(calls) == 1
|
||||
assert cache_module._cache_initialized is True
|
||||
|
||||
def test_closed_current_loop_falls_back_to_fresh_loop(self, cache_globals, monkeypatch, tmp_path):
|
||||
"""A closed loop left as the current loop must not be driven; fall back to ``asyncio.run``."""
|
||||
cfg = tmp_path / "extensions_config.json"
|
||||
_write_extensions_config(cfg, {"srv1": _server()})
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg))
|
||||
calls: list[str] = []
|
||||
|
||||
async def _fake_tools(**_kwargs):
|
||||
calls.append(threading.current_thread().name)
|
||||
return []
|
||||
|
||||
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
|
||||
closed_loop = asyncio.new_event_loop()
|
||||
closed_loop.close()
|
||||
asyncio.set_event_loop(closed_loop)
|
||||
try:
|
||||
assert cache_module.get_cached_mcp_tools() == []
|
||||
finally:
|
||||
asyncio.set_event_loop(None)
|
||||
|
||||
assert len(calls) == 1
|
||||
assert cache_module._cache_initialized is True
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user