mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-25 05:56:18 +00:00
fix(client): honor agent MCP plugin selections (#5630)
* fix(client): honor named-agent MCP plugin selections * fix(client): normalize MCP selection cache identity
This commit is contained in:
parent
29d285731b
commit
60d5659d1d
@ -1848,6 +1848,12 @@ DeerFlow is model-agnostic — it works with any LLM that implements the OpenAI-
|
|||||||
|
|
||||||
## Embedded Python Client
|
## Embedded Python Client
|
||||||
|
|
||||||
|
For `DeerFlowClient(agent_name="researcher")`, the named agent's `mcp_plugins`
|
||||||
|
selection applies to both the lead agent and its `task` / `batch_task`
|
||||||
|
delegations: `null` inherits all enabled MCP plugins, `[]` selects none, and
|
||||||
|
installation IDs select only those plugins. Call `client.reset_agent()` after
|
||||||
|
editing the saved agent configuration to refresh the selection.
|
||||||
|
|
||||||
`DeerFlowClient.stream()` includes `summary_text` in each `values` event. This is the current compacted context summary, or `None` when absent. Consumers can record changes without reading checkpoint internals; repeated snapshots may carry the same summary, and an initial snapshot may already contain one from an earlier turn.
|
`DeerFlowClient.stream()` includes `summary_text` in each `values` event. This is the current compacted context summary, or `None` when absent. Consumers can record changes without reading checkpoint internals; repeated snapshots may carry the same summary, and an initial snapshot may already contain one from an earlier turn.
|
||||||
|
|
||||||
DeerFlow can be used as an embedded Python library without running the full HTTP services. The `DeerFlowClient` provides direct in-process access to all agent and Gateway capabilities, returning the same response schemas as the HTTP Gateway API. The HTTP Gateway also exposes `DELETE /api/threads/{thread_id}` to remove DeerFlow-managed local thread data after the LangGraph thread itself has been deleted:
|
DeerFlow can be used as an embedded Python library without running the full HTTP services. The `DeerFlowClient` provides direct in-process access to all agent and Gateway capabilities, returning the same response schemas as the HTTP Gateway API. The HTTP Gateway also exposes `DELETE /api/threads/{thread_id}` to remove DeerFlow-managed local thread data after the LangGraph thread itself has been deleted:
|
||||||
|
|||||||
@ -62,10 +62,10 @@ drift.
|
|||||||
- `"custom"` — forwarded from `StreamWriter`; DeerFlow-built-in custom events are dual-emitted through `deerflow.utils.custom_events`, so `astream_events(version="v2")` consumers also receive one `on_custom_event` with `name=payload["type"]` and the unchanged payload as `data`
|
- `"custom"` — forwarded from `StreamWriter`; DeerFlow-built-in custom events are dual-emitted through `deerflow.utils.custom_events`, so `astream_events(version="v2")` consumers also receive one `on_custom_event` with `name=payload["type"]` and the unchanged payload as `data`
|
||||||
- `"end"` — stream finished (carries cumulative `usage` counted once per message id)
|
- `"end"` — stream finished (carries cumulative `usage` counted once per message id)
|
||||||
- **Custom-event invariant** — use `emit_custom_event` / `aemit_custom_event`, never `StreamWriter` alone. Built-in payloads require a non-empty string `type`; typeless payloads stay writer-only, absent from `astream_events`. The writer runs first and is authoritative for Gateway/Web UI/embedded clients; best-effort callbacks must not break it. Async graph hooks must await the async helper, never dispatch synchronously on a running event loop.
|
- **Custom-event invariant** — use `emit_custom_event` / `aemit_custom_event`, never `StreamWriter` alone. Built-in payloads require a non-empty string `type`; typeless payloads stay writer-only, absent from `astream_events`. The writer runs first and is authoritative for Gateway/Web UI/embedded clients; best-effort callbacks must not break it. Async graph hooks must await the async helper, never dispatch synchronously on a running event loop.
|
||||||
- Agent created lazily via `create_agent()` + `build_middlewares()`, same as `make_lead_agent`
|
- Lazy graph creation uses `create_agent()` + `build_middlewares()`.
|
||||||
- Cache graphs by effective storage `user_id` in every auth mode because prompts and middleware bind user SOUL, skills, and storage. `stream()` must materialize it before worker or isolated-loop boundaries.
|
- Cache graphs by storage `user_id` and the unordered set of named-agent `mcp_plugins`. `stream()` materializes `user_id` before worker/loop boundaries in every auth mode.
|
||||||
- Supports `checkpointer` parameter for state persistence across turns
|
- Supports `checkpointer` parameter for state persistence across turns
|
||||||
- `reset_agent()` forces agent recreation (e.g. after memory or skill changes)
|
- `reset_agent()` reloads AgentConfig and rebuilds the graph. Every run's metadata carries `mcp_plugins` for delegation, including cache hits.
|
||||||
- [Streaming design](../../../docs/STREAMING.md): Gateway/client parallel paths, LangGraph `stream_mode`, per-id deduplication, and regression tests
|
- [Streaming design](../../../docs/STREAMING.md): Gateway/client parallel paths, LangGraph `stream_mode`, per-id deduplication, and regression tests
|
||||||
|
|
||||||
**Gateway Equivalent Methods** (replaces Gateway API):
|
**Gateway Equivalent Methods** (replaces Gateway API):
|
||||||
|
|||||||
@ -324,6 +324,9 @@ class DeerFlowClient:
|
|||||||
self._loaded_agent_config_key = loaded_config_key
|
self._loaded_agent_config_key = loaded_config_key
|
||||||
self._loaded_agent_config = agent_config
|
self._loaded_agent_config = agent_config
|
||||||
memory_enabled = getattr(agent_config, "memory_enabled", True) is not False
|
memory_enabled = getattr(agent_config, "memory_enabled", True) is not False
|
||||||
|
mcp_plugins = getattr(agent_config, "mcp_plugins", None)
|
||||||
|
# Delegation reads this run's metadata, including when the graph is cached.
|
||||||
|
config.setdefault("metadata", {})["mcp_plugins"] = mcp_plugins
|
||||||
|
|
||||||
authorization_identity = None
|
authorization_identity = None
|
||||||
if self._app_config.authorization.enabled:
|
if self._app_config.authorization.enabled:
|
||||||
@ -349,6 +352,7 @@ class DeerFlowClient:
|
|||||||
cfg.get("max_total_subagents"),
|
cfg.get("max_total_subagents"),
|
||||||
self._agent_name,
|
self._agent_name,
|
||||||
memory_enabled,
|
memory_enabled,
|
||||||
|
frozenset(mcp_plugins) if mcp_plugins is not None else None,
|
||||||
frozenset(self._available_skills) if self._available_skills is not None else None,
|
frozenset(self._available_skills) if self._available_skills is not None else None,
|
||||||
self._checkpoint_channel_mode,
|
self._checkpoint_channel_mode,
|
||||||
self._checkpoint_snapshot_frequency,
|
self._checkpoint_snapshot_frequency,
|
||||||
@ -390,7 +394,7 @@ class DeerFlowClient:
|
|||||||
)
|
)
|
||||||
max_total_subagents = cfg.get("max_total_subagents", self._app_config.subagents.max_total_per_run)
|
max_total_subagents = cfg.get("max_total_subagents", self._app_config.subagents.max_total_per_run)
|
||||||
|
|
||||||
tools = self._get_tools(model_name=model_name, subagent_enabled=subagent_enabled)
|
tools = self._get_tools(model_name=model_name, subagent_enabled=subagent_enabled, mcp_plugins=mcp_plugins)
|
||||||
|
|
||||||
# Add framework-provided tools before authorization so Layer 1 sees
|
# Add framework-provided tools before authorization so Layer 1 sees
|
||||||
# every capability that can become model-visible.
|
# every capability that can become model-visible.
|
||||||
@ -483,11 +487,11 @@ class DeerFlowClient:
|
|||||||
logger.info("Agent created: agent_name=%s, model=%s, thinking=%s", self._agent_name, model_name, thinking_enabled)
|
logger.info("Agent created: agent_name=%s, model=%s, thinking=%s", self._agent_name, model_name, thinking_enabled)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _get_tools(*, model_name: str | None, subagent_enabled: bool):
|
def _get_tools(*, model_name: str | None, subagent_enabled: bool, mcp_plugins: list[str] | None = None):
|
||||||
"""Lazy import to avoid circular dependency at module level."""
|
"""Lazy import to avoid circular dependency at module level."""
|
||||||
from deerflow.tools import get_available_tools
|
from deerflow.tools import get_available_tools
|
||||||
|
|
||||||
return get_available_tools(model_name=model_name, subagent_enabled=subagent_enabled)
|
return get_available_tools(model_name=model_name, subagent_enabled=subagent_enabled, mcp_plugins=mcp_plugins)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _serialize_tool_calls(tool_calls) -> list[dict]:
|
def _serialize_tool_calls(tool_calls) -> list[dict]:
|
||||||
|
|||||||
@ -1277,6 +1277,99 @@ class TestExtractText:
|
|||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestClientMcpSelection:
|
||||||
|
@pytest.fixture
|
||||||
|
def mcp_client(self, client):
|
||||||
|
from deerflow.config.app_config import AppConfig
|
||||||
|
from deerflow.config.sandbox_config import SandboxConfig
|
||||||
|
|
||||||
|
app_config = AppConfig(models=[], sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"))
|
||||||
|
app_config.tool_search.enabled = False
|
||||||
|
client._app_config = app_config
|
||||||
|
client._agent_name = "researcher"
|
||||||
|
extensions = ExtensionsConfig.model_validate({"mcpServers": {name: {"enabled": True, "capability": {"id": identity}} for name, identity in [("work", "installation-A"), ("personal", "installation-B")]}})
|
||||||
|
cached_tools = [tag_mcp_tool(StructuredTool.from_function(lambda: "result", name=f"{name}_search", description="Search"), server_name=name) for name in extensions.mcp_servers]
|
||||||
|
graph = MagicMock()
|
||||||
|
graph.stream.return_value = []
|
||||||
|
with (
|
||||||
|
patch("deerflow.client.create_chat_model"),
|
||||||
|
patch("deerflow.client.create_agent", return_value=graph) as create_agent,
|
||||||
|
patch("deerflow.client.build_middlewares", return_value=[]),
|
||||||
|
patch("deerflow.client.apply_prompt_template", return_value="prompt"),
|
||||||
|
patch("deerflow.client.get_enabled_skills_for_config", return_value=[]),
|
||||||
|
patch("deerflow.client.load_agent_config") as load_config,
|
||||||
|
patch("deerflow.tools.tools.get_app_config", return_value=app_config),
|
||||||
|
patch("deerflow.config.acp_config.get_acp_agents", return_value={}),
|
||||||
|
patch.object(ExtensionsConfig, "from_file", return_value=extensions),
|
||||||
|
patch("deerflow.mcp.cache.get_cached_mcp_tools", return_value=cached_tools),
|
||||||
|
patch("deerflow.runtime.checkpointer.get_checkpointer", return_value=None),
|
||||||
|
):
|
||||||
|
yield SimpleNamespace(client=client, graph=graph, create_agent=create_agent, load_config=load_config, cached_tools=cached_tools)
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("selection", "expected_names"),
|
||||||
|
[(None, ["work_search", "personal_search"]), ([], []), (["installation-A"], ["work_search"])],
|
||||||
|
)
|
||||||
|
def test_selects_mcp_tools_without_changing_shared_cache(self, mcp_client, selection, expected_names):
|
||||||
|
from deerflow.tools.mcp_metadata import is_mcp_tool
|
||||||
|
|
||||||
|
mcp_client.load_config.return_value = AgentConfig(name="researcher", mcp_plugins=selection)
|
||||||
|
config = mcp_client.client._get_runnable_config("t1")
|
||||||
|
mcp_client.client._ensure_agent(config)
|
||||||
|
|
||||||
|
tools = mcp_client.create_agent.call_args.kwargs["tools"]
|
||||||
|
assert [tool.name for tool in tools if is_mcp_tool(tool)] == expected_names
|
||||||
|
assert [tool.name for tool in mcp_client.cached_tools] == ["work_search", "personal_search"]
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("selection", [None, [], ["installation-A"]])
|
||||||
|
def test_each_stream_carries_mcp_selection_for_delegation_on_cache_hit(self, mcp_client, selection):
|
||||||
|
mcp_client.load_config.return_value = AgentConfig(name="researcher", mcp_plugins=selection)
|
||||||
|
for _ in range(2):
|
||||||
|
list(mcp_client.client.stream("hello", thread_id="t1"))
|
||||||
|
|
||||||
|
mcp_client.create_agent.assert_called_once()
|
||||||
|
mcp_client.load_config.assert_called_once()
|
||||||
|
assert mcp_client.graph.stream.call_count == 2
|
||||||
|
for call in mcp_client.graph.stream.call_args_list:
|
||||||
|
metadata = call.kwargs["config"]["metadata"]
|
||||||
|
assert metadata["mcp_plugins"] == selection
|
||||||
|
|
||||||
|
def test_reuses_graph_when_mcp_selection_order_changes(self, mcp_client):
|
||||||
|
agent_config = AgentConfig(name="researcher", mcp_plugins=["installation-A", "installation-B"])
|
||||||
|
mcp_client.load_config.return_value = agent_config
|
||||||
|
client = mcp_client.client
|
||||||
|
client._ensure_agent(client._get_runnable_config("t1"))
|
||||||
|
|
||||||
|
agent_config.mcp_plugins = ["installation-B", "installation-A"]
|
||||||
|
config = client._get_runnable_config("t2")
|
||||||
|
client._ensure_agent(config)
|
||||||
|
|
||||||
|
mcp_client.create_agent.assert_called_once()
|
||||||
|
assert config["metadata"]["mcp_plugins"] == ["installation-B", "installation-A"]
|
||||||
|
|
||||||
|
def test_reset_refreshes_mcp_selection_and_graph_cache_identity(self, mcp_client):
|
||||||
|
client = mcp_client.client
|
||||||
|
keys = []
|
||||||
|
for selection in [None, [], ["installation-A"]]:
|
||||||
|
mcp_client.load_config.return_value = AgentConfig(name="researcher", mcp_plugins=selection)
|
||||||
|
client.reset_agent()
|
||||||
|
config = client._get_runnable_config("t1")
|
||||||
|
config["metadata"] = {"existing": "preserved", "mcp_plugins": ["installation-B"]}
|
||||||
|
client._ensure_agent(config)
|
||||||
|
keys.append(client._agent_config_key)
|
||||||
|
assert config["metadata"] == {"existing": "preserved", "mcp_plugins": selection}
|
||||||
|
|
||||||
|
# Changing the saved config takes effect only after reset_agent().
|
||||||
|
mcp_client.load_config.return_value = AgentConfig(name="researcher", mcp_plugins=["installation-B"])
|
||||||
|
cached_config = client._get_runnable_config("t2")
|
||||||
|
client._ensure_agent(cached_config)
|
||||||
|
assert cached_config["metadata"]["mcp_plugins"] == selection
|
||||||
|
|
||||||
|
assert keys[0] != keys[1] != keys[2]
|
||||||
|
assert mcp_client.load_config.call_count == 3
|
||||||
|
assert mcp_client.create_agent.call_count == 3
|
||||||
|
|
||||||
|
|
||||||
class TestEnsureAgent:
|
class TestEnsureAgent:
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
("agent_name", "agent_config", "expected_memory_enabled"),
|
("agent_name", "agent_config", "expected_memory_enabled"),
|
||||||
@ -1691,6 +1784,7 @@ class TestEnsureAgent:
|
|||||||
None,
|
None,
|
||||||
True,
|
True,
|
||||||
None,
|
None,
|
||||||
|
None,
|
||||||
"full",
|
"full",
|
||||||
10,
|
10,
|
||||||
get_effective_user_id(),
|
get_effective_user_id(),
|
||||||
|
|||||||
@ -679,7 +679,7 @@ def _stub_client_assembly(monkeypatch) -> dict[str, str]:
|
|||||||
)
|
)
|
||||||
monkeypatch.setattr("deerflow.client.create_agent", lambda **kwargs: object())
|
monkeypatch.setattr("deerflow.client.create_agent", lambda **kwargs: object())
|
||||||
monkeypatch.setattr("deerflow.client.build_middlewares", lambda *args, **kwargs: [])
|
monkeypatch.setattr("deerflow.client.build_middlewares", lambda *args, **kwargs: [])
|
||||||
monkeypatch.setattr("deerflow.client.DeerFlowClient._get_tools", staticmethod(lambda *, model_name, subagent_enabled: [])) # noqa: ARG005
|
monkeypatch.setattr("deerflow.client.DeerFlowClient._get_tools", staticmethod(lambda *, model_name, subagent_enabled, mcp_plugins=None: [])) # noqa: ARG005
|
||||||
monkeypatch.setattr("deerflow.client.get_enabled_skills_for_config", lambda app_config: []) # noqa: ARG005
|
monkeypatch.setattr("deerflow.client.get_enabled_skills_for_config", lambda app_config: []) # noqa: ARG005
|
||||||
monkeypatch.setattr(
|
monkeypatch.setattr(
|
||||||
"deerflow.client.build_skill_search_setup",
|
"deerflow.client.build_skill_search_setup",
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user