fix(builder): ground installed MCP tools in native catalogue

api/core/tools/mcp_tool/provider.py

@@ -9,6 +9,7 @@ from core.tools.entities.tool_entities import (
     ToolDescription,
     ToolEntity,
     ToolIdentity,
+    ToolParameter,
     ToolProviderEntityWithPlugin,
     ToolProviderIdentity,
     ToolProviderType,
@@ -81,7 +82,7 @@ class MCPToolProviderController(ToolProviderController[ToolProviderEntityWithPlu
                     provider=entity.server_identifier,
                     icon=entity.icon if isinstance(entity.icon, str) else "",
                 ),
-                parameters=ToolTransformService.convert_mcp_schema_to_parameter(remote_mcp_tool.inputSchema),
+                parameters=cls.parameters_from_schema(remote_mcp_tool.inputSchema),
                 description=ToolDescription(
                     human=I18nObject(
                         en_US=remote_mcp_tool.description or "", zh_Hans=remote_mcp_tool.description or ""
@@ -117,6 +118,15 @@ class MCPToolProviderController(ToolProviderController[ToolProviderEntityWithPlu
             identity_mode=entity.identity_mode,
         )
 
+    @staticmethod
+    def parameters_from_schema(schema: dict[str, Any]) -> list[ToolParameter]:
+        """Convert a public cached input declaration without constructing a provider.
+
+        Uses the ordinary native MCP mapping, with no credentials, sessions or
+        transport. Invalid schemas retain the native converter's failure behavior.
+        """
+        return ToolTransformService.convert_mcp_schema_to_parameter(schema)
+
     def _validate_credentials(self, user_id: str, credentials: dict[str, Any]):
         """
         validate the credentials of the provider

api/core/workflow/generator/tool_catalogue.py

@@ -2,7 +2,7 @@
 Tool catalogue for the workflow generator.
 
 Returns a compact, LLM-readable inventory of the tools currently installed for
-a tenant (hardcoded built-in, plugin and native API providers). The planner
+a tenant (hardcoded built-in, plugin, native API and cached MCP providers). The planner
 uses this to recommend ``tool`` nodes by exact ``provider/tool`` identifier;
 the builder consumes the same list so it can emit a syntactically correct
 ``tool`` node ``data`` block (provider_id, provider_type, tool_name,
@@ -34,6 +34,7 @@ from core.tools.entities.tool_entities import ToolDescription
 from core.tools.plugin_tool.provider import PluginToolProviderController
 from core.tools.tool_manager import ToolManager
 from repositories.api_tool_catalogue import list_api_tool_metadata
+from repositories.mcp_tool_catalogue import list_mcp_tool_metadata
 
 logger = logging.getLogger(__name__)
 
@@ -79,7 +80,8 @@ class ToolCatalogueEntry(TypedDict):
     tool_label: str
     description: str  # one-line LLM-friendly description
     # Builtin/plugin: provider_controller.need_credentials. API: conservative
-    # unknown-auth flag; authorization_metadata must take precedence.
+    # unknown-auth flag; authorization_metadata must take precedence. MCP
+    # configuration uses mcp_cached_auth, never builtin credential ownership.
     needs_credentials: bool
     authorization_metadata: NotRequired[str]
     # Provider declaration metadata, searched with a low weight so a tool whose
@@ -88,7 +90,10 @@ class ToolCatalogueEntry(TypedDict):
     provider_description: NotRequired[str]
     provider_tags: NotRequired[list[str]]
     parameters: NotRequired[list[dict[str, Any]]]
+    input_schema: NotRequired[dict[str, Any]]
     output_schema: NotRequired[dict[str, Any]]
+    # Cached native MCP configuration only; never evidence of runtime access.
+    mcp_cached_auth: NotRequired[bool]
     # Direct public HTTP response declarations, distinct from native outputs.
     response_contracts: NotRequired[dict[str, object]]
 
@@ -127,8 +132,8 @@ def build_tool_catalogue(
     Failures inside a single provider (mis-declared tool, plugin runtime
     error) are logged and skipped — one bad provider must not break the
     whole generator. A failed plugin inventory fetch is logged and leaves the
-    hardcoded built-in tools in place. Native API public metadata is read
-    independently and never constructs credential-bearing controllers.
+    hardcoded built-in tools in place. Native API and cached MCP public metadata
+    are read independently without constructing credential-bearing controllers.
     Returns the complete installed
     inventory; prompt-specific truncation belongs to ``format_tool_catalogue``.
     """
@@ -213,6 +218,13 @@ def build_tool_catalogue(
             on_source_error("api_tool_inventory")
         logger.exception("Workflow generator: failed to list public API tool metadata; keeping other tools")
 
+    try:
+        entries.extend(list_mcp_tool_metadata(tenant_id))
+    except Exception:
+        if on_source_error is not None:
+            on_source_error("mcp_tool_inventory")
+        logger.exception("Workflow generator: failed to list public cached MCP metadata; keeping other tools")
+
     entries.sort(key=itemgetter("provider_name", "tool_name"))
     return entries
 
@@ -258,8 +270,8 @@ def format_tool_catalogue(entries: list[ToolCatalogueEntry], *, max_tools: int |
             f" [provider_id={json.dumps(e['provider_name'], ensure_ascii=False)}; "
             f"tool_name={json.dumps(e['tool_name'], ensure_ascii=False)}"
         )
-        if e["provider_type"] == "api":
-            identity_fields += '; provider_type="api"'
+        if e["provider_type"] in {"api", "mcp"}:
+            identity_fields += f"; provider_type={json.dumps(e['provider_type'])}"
         line += identity_fields + "]"
         if desc:
             line += f" — {desc}"
@@ -601,6 +613,9 @@ def format_tool_builder_context(entry: ToolCatalogueEntry) -> str:
             suffix = " — " + "; ".join(suffixes) if suffixes else ""
             lines.append(f"- {name}: {type_}, form={form}, {requirement}{suffix}")
 
+    if "input_schema" in entry:
+        lines.extend(["", "Declared MCP input schema:", json.dumps(entry["input_schema"], ensure_ascii=False)])
+
     lines.extend(
         [
             "",

api/repositories/mcp_tool_catalogue.py

@@ -0,0 +1,88 @@
+"""Tenant-scoped cached MCP declarations, without secret or network access.
+
+The native MCP schema converter owns parameter mapping. Discovery selects only
+public columns, preserves the server identifier used by ToolManager, and treats
+cached authentication as configuration evidence, never live runtime success.
+"""
+
+import json
+import logging
+from typing import TYPE_CHECKING
+
+from jsonschema import Draft202012Validator, SchemaError
+from sqlalchemy import select
+
+from core.db.session_factory import session_factory
+from core.mcp.types import Tool as RemoteMCPTool
+from core.tools.mcp_tool.provider import MCPToolProviderController
+from models.tools import MCPToolProvider
+
+if TYPE_CHECKING:
+    from core.workflow.generator.tool_catalogue import ToolCatalogueEntry
+
+logger = logging.getLogger(__name__)
+
+
+def list_mcp_tool_metadata(tenant_id: str) -> list["ToolCatalogueEntry"]:
+    """Return owned public cached actions; malformed providers/actions are omitted.
+
+    Requires an admitted workspace identity. Every query is tenant scoped and
+    reads no URL, header, credential or creator columns. No remote refresh,
+    decryption or authorization occurs; source query failures propagate for the
+    caller to report an incomplete inventory. MCP's native owner exposes these
+    providers workspace-wide, independently of builtin credential visibility.
+    """
+    statement = select(
+        MCPToolProvider.tenant_id,
+        MCPToolProvider.server_identifier,
+        MCPToolProvider.name,
+        MCPToolProvider.tools,
+        MCPToolProvider.authed,
+    ).where(MCPToolProvider.tenant_id == tenant_id)
+    with session_factory.create_session() as session:
+        rows = session.execute(statement).mappings().all()
+
+    entries: list[ToolCatalogueEntry] = []
+    for row in rows:
+        try:
+            server = row["server_identifier"]
+            if row["tenant_id"] != tenant_id or not isinstance(server, str) or not server.strip():
+                raise ValueError("Invalid MCP provider identity")
+            if not isinstance(row["name"], str):
+                raise ValueError("Invalid MCP provider label")
+            tools = json.loads(row["tools"])
+            if not isinstance(tools, list):
+                raise ValueError("Invalid cached MCP declarations")
+        except (TypeError, ValueError):
+            logger.warning("Workflow generator: invalid public MCP provider metadata; skipping provider")
+        else:
+            for raw_tool in tools:
+                try:
+                    tool = RemoteMCPTool.model_validate(raw_tool)
+                    if not tool.name.strip():
+                        raise ValueError("Invalid MCP action identity")
+                    Draft202012Validator.check_schema(tool.inputSchema)
+                    if tool.outputSchema is not None:
+                        Draft202012Validator.check_schema(tool.outputSchema)
+                    parameters = MCPToolProviderController.parameters_from_schema(tool.inputSchema)
+                    entries.append(
+                        {
+                            "provider_name": server,
+                            "provider_type": "mcp",
+                            "plugin_id": "",
+                            "provider_label": row["name"],
+                            "tool_name": tool.name,
+                            "tool_label": tool.title or tool.name,
+                            "description": tool.description or "",
+                            "parameters": [parameter.model_dump(mode="json") for parameter in parameters],
+                            "input_schema": tool.inputSchema,
+                            "output_schema": tool.outputSchema or {},
+                            "needs_credentials": True,
+                            "mcp_cached_auth": row["authed"] is True,
+                        }
+                    )
+                except (TypeError, ValueError, AttributeError, SchemaError):
+                    # Do not log raw declarations or validation exceptions: a
+                    # malformed cached declaration may contain arbitrary text.
+                    logger.warning("Workflow generator: invalid public MCP action metadata; skipping action")
+    return entries

api/services/dify_builder/agent/resources.py

@@ -301,6 +301,10 @@ def _tool_readiness(entry: ToolCatalogueEntry, authorized: set[str] | None) -> t
     with a note saying it could not be verified. Runtime authorization
     remains the final gate either way.
     """
+    if entry.get("provider_type") == "mcp":
+        # Native MCP configuration is workspace-scoped, not a builtin credential
+        # row. Cached auth does not establish current remote authorization.
+        return ("ready", "") if entry.get("mcp_cached_auth") else ("missing_config", SETUP_REQUIRED_NOTE)
     # API authorization belongs to its native owner, never builtin credential
     # rows. Public metadata cannot prove authorization even for an API tool
     # that uses no credentials; keep it bindable with an honest setup note.
@@ -357,6 +361,7 @@ def _tool_ref(entry: ToolCatalogueEntry, authorized: set[str] | None) -> Resourc
         else None,
         declaration_digest=canonical_digest(
             {
+                **({"input_schema": entry["input_schema"]} if "input_schema" in entry else {}),
                 "parameters": entry.get("parameters", []),
                 "output_schema": entry.get("output_schema", {}),
             }
@@ -450,7 +455,7 @@ def list_tenant_resources(
         authorized: set[str] | None = None
         if any(
             t.get("needs_credentials")
-            and t.get("provider_type") != "api"
+            and t.get("provider_type") not in {"api", "mcp"}
             and t.get("authorization_metadata") != "unverified"
             for t in tools
         ):

api/services/dify_builder/effect_disclosure.py

@@ -609,6 +609,10 @@ def _tool(data: Mapping[str, Any]) -> tuple[EffectTier, str, str]:
         return "destructive", target, obj
     if words & _WRITE_WORDS:
         return "side_effect", target, obj
+    if provider_type == "mcp":
+        # Remote names and cached annotations cannot authorize opaque actions.
+        # Retain explicit confirmation even for names such as docs_search.
+        return "side_effect", target, obj
     if words & _READ_WORDS or provider_type == "dataset-retrieval":
         return "read", target, obj
     return "side_effect", target, obj

api/tests/unit_tests/services/tools/test_mcp_tool_catalogue.py

@@ -0,0 +1,476 @@
+"""Installed cached MCP definitions reach Builder and the ordinary native Tool adapter.
+
+The database and schema converter are real. Discovery forbids network/decryption;
+native execution uses a recording local transport, never a live MCP service.
+"""
+
+import json
+from copy import deepcopy
+from typing import NoReturn, NotRequired, TypedDict
+from uuid import uuid4
+
+import pytest
+from sqlalchemy import delete, insert, update
+from sqlalchemy.engine import Engine
+
+from core.tools.tool_manager import ToolManager
+from core.workflow.generator.tool_catalogue import (
+    ToolCatalogueEntry,
+    build_tool_catalogue,
+    find_tool_entry,
+    format_tool_builder_context,
+    format_tool_catalogue,
+)
+from core.workflow.generator.types import GraphDict
+from models.tools import MCPToolProvider
+from services.dify_builder.agent import resources
+from services.dify_builder.effect_disclosure import disclose
+from services.dify_builder.resource_grounding import (
+    ResourceBindingError,
+    resource_fingerprint,
+    validate_changed_resources,
+)
+from tests.unit_tests.services.dify_builder.installed_tool_inventory import (
+    ACCOUNT_A,
+    ACCOUNT_B,
+    OTHER_TENANT_ID,
+    TENANT_ID,
+    FakeToolDaemon,
+    bind_database,
+    install_fake_daemon,
+    record_sql,
+)
+
+SERVER = "owned-microsoft-docs"
+ACTION = "microsoft_docs_search"
+OUTPUT: dict[str, object] = {
+    "type": "object",
+    "properties": {"results": {"type": "array", "items": {"type": "string"}}},
+}
+
+
+class CachedDeclaration(TypedDict):
+    name: str
+    title: str
+    description: str
+    inputSchema: dict[str, object]
+    outputSchema: NotRequired[dict[str, object]]
+
+
+def definition() -> CachedDeclaration:
+    return {
+        "name": ACTION,
+        "title": "Microsoft documentation search",
+        "description": "Search Microsoft documentation using the user's question.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {"query": {"type": "string", "description": "The user's documentation question."}},
+            "required": ["query"],
+        },
+        "outputSchema": deepcopy(OUTPUT),
+    }
+
+
+def add_provider(engine: Engine, *, tenant_id: str = TENANT_ID, authed: bool = True, tools: object = None) -> None:
+    with engine.begin() as connection:
+        connection.execute(
+            insert(MCPToolProvider).values(
+                id=str(uuid4()),
+                tenant_id=tenant_id,
+                user_id=ACCOUNT_B,
+                name="Microsoft Docs (display only)",
+                server_identifier=SERVER,
+                server_url="encrypted-url-marker",
+                server_url_hash="synthetic-hash",
+                encrypted_headers='{"secret":"encrypted-header-marker"}',
+                encrypted_credentials='{"secret":"encrypted-credential-marker"}',
+                icon="docs.svg",
+                authed=authed,
+                tools=json.dumps([definition()] if tools is None else tools),
+            )
+        )
+
+
+@pytest.fixture
+def inventory(monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> Engine:
+    from core.entities.mcp_provider import MCPProviderEntity
+    from services.tools.mcp_tools_manage_service import MCPToolManageService
+
+    bind_database(monkeypatch, sqlite_engine)
+    install_fake_daemon(monkeypatch, FakeToolDaemon())
+    monkeypatch.setattr(resources, "_list_models", lambda _tenant: [])
+    monkeypatch.setattr(resources, "_list_datasets", lambda _tenant, _account: [])
+
+    def forbidden(*_args: object, **_kwargs: object) -> NoReturn:
+        raise AssertionError("Discovery cannot decrypt, resolve credentials or contact MCP")
+
+    for name in ("decrypt_server_url", "decrypt_headers", "decrypt_credentials", "retrieve_tokens"):
+        monkeypatch.setattr(MCPProviderEntity, name, forbidden)
+    monkeypatch.setattr(MCPToolManageService, "list_provider_tools", forbidden)
+    monkeypatch.setattr(ToolManager, "get_mcp_provider_controller", forbidden)
+    return sqlite_engine
+
+
+def test_cached_mcp_discovery_reads_only_public_columns_and_exact_native_identity(inventory: Engine) -> None:
+    add_provider(inventory)
+    statements = record_sql(inventory)
+    entries = build_tool_catalogue(TENANT_ID)
+    entry = find_tool_entry(entries, SERVER, ACTION)
+    assert entry is not None
+    assert entry["provider_type"] == "mcp"
+    assert entry["provider_label"] == "Microsoft Docs (display only)"
+    assert entry["tool_label"] == "Microsoft documentation search"
+    assert entry["output_schema"] == OUTPUT
+    assert entry["input_schema"] == definition()["inputSchema"]
+    assert json.dumps(definition()["inputSchema"]) in format_tool_builder_context(entry)
+    assert [(p["name"], p["type"], p["required"]) for p in entry["parameters"]] == [("query", "string", True)]
+    assert entry["parameters"][0]["llm_description"] == "The user's documentation question."
+    prompt = format_tool_catalogue([entry])
+    assert 'provider_type="mcp"' in prompt
+    assert f'provider_id="{SERVER}"' in prompt
+    queries = [sql.lower() for sql in statements if "tool_mcp_providers" in sql.lower()]
+    assert queries
+    assert all("tenant_id =" in sql for sql in queries)
+    assert all(
+        not any(secret in sql for secret in ("server_url", "encrypted_headers", "encrypted_credentials"))
+        for sql in queries
+    )
+    assert "encrypted-" not in repr(entry)
+
+
+def test_mcp_configuration_does_not_borrow_builtin_credentials(
+    inventory: Engine, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    add_provider(inventory)
+
+    def forbidden(*_args: object) -> NoReturn:
+        raise AssertionError("MCP cannot use builtin credential visibility")
+
+    monkeypatch.setattr(resources, "_authorized_providers", forbidden)
+    refs = resources.list_tenant_resources(TENANT_ID).tools
+    ref = next(ref for ref in refs if ref.id == f"{SERVER}/{ACTION}")
+    assert ref.readiness == "ready"
+    assert ref.binding["provider_type"] == "mcp"
+    assert ref.availability["runtime"] == "unverified"
+    with inventory.begin() as connection:
+        connection.execute(update(MCPToolProvider).values(authed=False))
+    current = resources.list_tenant_resources(TENANT_ID)
+    revoked = next(item for item in current.tools if item.id == ref.id)
+    assert revoked.readiness == "missing_config"
+    assert resource_fingerprint(revoked) != resource_fingerprint(ref)
+
+
+@pytest.mark.parametrize("mutation", ["foreign", "deleted", "schema", "input_schema", "revoked"])
+def test_selected_mcp_binding_rejects_stale_or_foreign_declarations(inventory: Engine, mutation: str) -> None:
+    add_provider(inventory)
+    catalogue = resources.list_tenant_resources(TENANT_ID)
+    ref = next((item for item in catalogue.tools if item.id == f"{SERVER}/{ACTION}"), None)
+    assert ref is not None
+    graph = bound_graph(build_tool_catalogue(TENANT_ID))
+    with inventory.begin() as connection:
+        if mutation == "foreign":
+            connection.execute(update(MCPToolProvider).values(tenant_id=OTHER_TENANT_ID))
+        elif mutation == "deleted":
+            connection.execute(delete(MCPToolProvider))
+        elif mutation == "revoked":
+            connection.execute(update(MCPToolProvider).values(authed=False))
+        else:
+            changed = definition()
+            if mutation == "input_schema":
+                changed["inputSchema"]["additionalProperties"] = False
+            else:
+                changed["outputSchema"] = {"type": "object", "properties": {"unexpected": {"type": "string"}}}
+            connection.execute(update(MCPToolProvider).values(tools=json.dumps([changed])))
+    with pytest.raises(ResourceBindingError) as caught:
+        validate_changed_resources(
+            {"nodes": []},
+            dict(graph),
+            resources.list_tenant_resources(TENANT_ID),
+            confirmed_ids=[f"tool:{ref.id}"],
+            expected_fingerprints={f"tool:{ref.id}": resource_fingerprint(ref)},
+        )
+    assert caught.value.reason_code == (
+        "resource_not_in_catalogue" if mutation in {"foreign", "deleted"} else "resource_catalogue_changed"
+    )
+
+
+def bound_graph(entries: list[ToolCatalogueEntry]) -> GraphDict:
+    from services.workflow_generation.runner import WorkflowGenerator
+
+    graph: GraphDict = {
+        "nodes": [
+            {
+                "id": "start",
+                "type": "custom",
+                "position": {"x": 0, "y": 0},
+                "data": {
+                    "type": "start",
+                    "title": "Start",
+                    "variables": [
+                        {
+                            "variable": "question",
+                            "label": "Question",
+                            "type": "text-input",
+                            "required": True,
+                            "max_length": 200,
+                        }
+                    ],
+                },
+            },
+            {
+                "id": "search",
+                "type": "custom",
+                "position": {"x": 300, "y": 0},
+                "data": {
+                    "type": "tool",
+                    "title": "Search docs",
+                    "tool_configurations": {},
+                    "tool_parameters": {"query": {"type": "mixed", "value": "{{#start.question#}}"}},
+                },
+            },
+            {
+                "id": "end",
+                "type": "custom",
+                "position": {"x": 600, "y": 0},
+                "data": {
+                    "type": "end",
+                    "title": "End",
+                    "outputs": [{"variable": "results", "value_selector": ["search", "results"]}],
+                },
+            },
+        ],
+        "edges": [
+            {"id": "start-search", "type": "custom", "source": "start", "target": "search"},
+            {"id": "search-end", "type": "custom", "source": "search", "target": "end"},
+        ],
+        "viewport": {"x": 0, "y": 0, "zoom": 1},
+    }
+    WorkflowGenerator._hydrate_tool_nodes(
+        graph=graph,
+        plan_nodes=[{"id": "search", "node_type": "tool", "tool": {"provider_id": SERVER, "tool_name": ACTION}}],
+        tool_catalogue_entries=entries,
+    )
+    return graph
+
+
+def test_mcp_catalogue_constructs_bound_graph_without_schema_or_identity_guesses(inventory: Engine) -> None:
+    add_provider(inventory)
+    graph = bound_graph(build_tool_catalogue(TENANT_ID))
+    data = graph["nodes"][1]["data"]
+    assert data["provider_type"] == "mcp"
+    assert data["provider_id"] == SERVER
+    assert data["tool_name"] == ACTION
+    assert data["output_schema"] == OUTPUT
+    assert data["tool_parameters"]["query"] == {"type": "mixed", "value": "{{#start.question#}}"}
+    validate_changed_resources({"nodes": []}, dict(graph), resources.list_tenant_resources(TENANT_ID))
+    data["tool_parameters"] = {"invented": {"type": "constant", "value": "wrong"}}
+    with pytest.raises(ResourceBindingError, match="resource_configuration_incompatible"):
+        validate_changed_resources({"nodes": []}, dict(graph), resources.list_tenant_resources(TENANT_ID))
+
+
+def test_search_named_mcp_still_requires_explicit_confirmation() -> None:
+    graph = {
+        "nodes": [
+            {
+                "id": "search",
+                "data": {"type": "tool", "provider_type": "mcp", "provider_id": SERVER, "tool_name": ACTION},
+            }
+        ]
+    }
+    disclosure = disclose(graph)
+    assert disclosure.items[0].tier == "side_effect"
+    assert disclosure.mocked_node_ids({}) == frozenset({"search"})
+    assert disclosure.mocked_node_ids(disclosure.confirmation(["search"])) == frozenset()
+
+
+def test_bound_mcp_graph_uses_native_tool_adapter_and_end_returns_transport_output(
+    sqlite_engine: Engine,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import time
+    from typing import Self
+
+    import core.tools.mcp_tool.tool as mcp_module
+    from core.entities.mcp_provider import MCPProviderEntity
+    from core.mcp.types import CallToolResult, TextContent
+    from extensions import ext_database
+    from graphon.enums import WorkflowNodeExecutionStatus
+    from graphon.graph.graph import Graph
+    from graphon.node_events import StreamCompletedEvent
+    from graphon.nodes.end.end_node import EndNode
+    from graphon.nodes.start.start_node import StartNode
+    from graphon.nodes.tool.tool_node import ToolNode
+    from graphon.runtime import RuntimeState, VariablePool
+    from services.dify_builder.agent.build import bind_resources
+    from services.workflow_runtime.node_factory import DifyNodeFactory
+    from tests.unit_tests.services.dify_builder.installed_tool_inventory import InventoryDatabase
+    from tests.workflow_test_utils import build_test_graph_init_params
+
+    bind_database(monkeypatch, sqlite_engine)
+    monkeypatch.setattr(ext_database, "db", InventoryDatabase(sqlite_engine))
+    install_fake_daemon(monkeypatch, FakeToolDaemon())
+    monkeypatch.setattr(resources, "_list_models", lambda _tenant: [])
+    monkeypatch.setattr(resources, "_list_datasets", lambda _tenant, _account: [])
+    add_provider(sqlite_engine)
+    plan = bind_resources(None, TENANT_ID, ["Search Microsoft documentation"], [f"{SERVER}/{ACTION}"])
+    assert f"[{SERVER}/{ACTION}]" in plan[0]
+    graph_config = bound_graph(build_tool_catalogue(TENANT_ID))
+    calls: list[tuple[str, dict[str, object]]] = []
+
+    class RecordingMCPTransport:
+        def __init__(self, *, server_url: str, headers: dict[str, str], **_kwargs: object) -> None:
+            assert server_url == "https://controlled-transport.invalid/mcp"
+            assert headers == {}
+
+        def __enter__(self) -> Self:
+            return self
+
+        def __exit__(self, *_args: object) -> None:
+            pass
+
+        def invoke_tool(self, *, tool_name: str, tool_args: dict[str, object]) -> CallToolResult:
+            calls.append((tool_name, tool_args))
+            return CallToolResult(
+                content=[TextContent(type="text", text="Controlled documentation match")],
+                structuredContent={"results": ["Controlled result from native MCP transport"]},
+            )
+
+    monkeypatch.setattr(mcp_module, "MCPClientWithAuthRetry", RecordingMCPTransport)
+    monkeypatch.setattr(
+        MCPProviderEntity, "decrypt_server_url", lambda _self: "https://controlled-transport.invalid/mcp"
+    )
+    monkeypatch.setattr(MCPProviderEntity, "decrypt_headers", lambda _self: {})
+    monkeypatch.setattr(MCPProviderEntity, "retrieve_tokens", lambda _self: None)
+    pool = VariablePool()
+    pool.add(["start", "question"], "How does Azure work?")
+    state = RuntimeState(workflow_id="controlled-workflow", variable_pool=pool, start_at=time.perf_counter())
+    init = build_test_graph_init_params(tenant_id=TENANT_ID, user_id=ACCOUNT_A, graph_config=graph_config)
+    factory = DifyNodeFactory(init_params=init, runtime_state=state)
+    graph = Graph.init(graph_config=graph_config, node_factory=factory, root_node_id="start")
+    start_node = graph.nodes["start"]
+    assert isinstance(start_node, StartNode)
+    start = start_node._run()
+    assert start.outputs["question"] == "How does Azure work?"
+    search = graph.nodes["search"]
+    assert isinstance(search, ToolNode)
+    search.bind_execution_id("controlled-search-execution")
+    completed = [event.node_run_result for event in search._run() if isinstance(event, StreamCompletedEvent)]
+    assert len(completed) == 1
+    assert completed[0].status == WorkflowNodeExecutionStatus.SUCCEEDED, completed[0].error
+    assert calls == [(ACTION, {"query": "How does Azure work?"})]
+    assert completed[0].outputs["results"] == ["Controlled result from native MCP transport"]
+    for name, value in completed[0].outputs.items():
+        pool.add(["search", name], value)
+    end_node = graph.nodes["end"]
+    assert isinstance(end_node, EndNode)
+    end = end_node._run()
+    assert end.outputs == {"results": ["Controlled result from native MCP transport"]}
+
+
+@pytest.mark.parametrize(
+    "bad_tools", [None, [], [{"name": "bad", "inputSchema": {"type": "object", "properties": []}}]]
+)
+def test_empty_or_malformed_mcp_cache_never_fabricates_tools(inventory: Engine, bad_tools: object) -> None:
+    add_tools = "null" if bad_tools is None else json.dumps(bad_tools)
+    add_provider(inventory, tools=[])
+    with inventory.begin() as connection:
+        connection.execute(update(MCPToolProvider).values(tools=add_tools))
+    assert all(entry["provider_type"] != "mcp" for entry in build_tool_catalogue(TENANT_ID))
+
+
+@pytest.mark.usefixtures("inventory")
+def test_mcp_inventory_failure_is_reported_without_losing_other_tool_sources(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import core.workflow.generator.tool_catalogue as catalogue_module
+
+    def unavailable(_tenant: str) -> NoReturn:
+        raise RuntimeError("controlled metadata source failure")
+
+    monkeypatch.setattr(catalogue_module, "list_mcp_tool_metadata", unavailable)
+    failures: list[str] = []
+    entries = build_tool_catalogue(TENANT_ID, on_source_error=failures.append)
+    assert failures == ["mcp_tool_inventory"]
+    assert any(entry["provider_name"] == "time" for entry in entries)
+    current = resources.list_tenant_resources(TENANT_ID)
+    assert current.tool_source_error == "inventory_unavailable"
+
+
+@pytest.mark.parametrize("mutation", ["foreign", "deleted", "revoked"])
+def test_native_mcp_adapter_rechecks_current_owned_provider_before_dispatch(
+    sqlite_engine: Engine,
+    monkeypatch: pytest.MonkeyPatch,
+    mutation: str,
+) -> None:
+    from core.workflow.node_runtime import DifyToolNodeRuntime
+    from graphon.nodes.tool.entities import ToolNodeData
+    from graphon.nodes.tool.exc import ToolRuntimeResolutionError
+    from graphon.runtime import VariablePool
+    from tests.workflow_test_utils import build_test_run_context
+
+    bind_database(monkeypatch, sqlite_engine)
+    add_provider(sqlite_engine)
+    runtime = DifyToolNodeRuntime(build_test_run_context(tenant_id=TENANT_ID, user_id=ACCOUNT_A))
+    node_data = ToolNodeData.model_validate(
+        {
+            "type": "tool",
+            "title": "Search docs",
+            "provider_id": SERVER,
+            "provider_name": SERVER,
+            "provider_type": "mcp",
+            "tool_name": ACTION,
+            "tool_label": "Search docs",
+            "tool_parameters": {},
+            "tool_configurations": {},
+        }
+    )
+    # The ordinary native owner can initially resolve this exact installed action.
+    runtime.get_runtime(node_id="search", node_data=node_data, variable_pool=VariablePool())
+    with sqlite_engine.begin() as connection:
+        if mutation == "foreign":
+            connection.execute(update(MCPToolProvider).values(tenant_id=OTHER_TENANT_ID))
+        elif mutation == "deleted":
+            connection.execute(delete(MCPToolProvider))
+        else:
+            # Native revoke clears cached actions as well as cached auth state.
+            connection.execute(update(MCPToolProvider).values(authed=False, tools="[]"))
+    with pytest.raises(ToolRuntimeResolutionError):
+        runtime.get_runtime(node_id="search", node_data=node_data, variable_pool=VariablePool())
+
+
+def test_mcp_without_output_declaration_keeps_native_generic_outputs(inventory: Engine) -> None:
+    tool = definition()
+    tool.pop("outputSchema")
+    add_provider(inventory, tools=[tool])
+    entry = find_tool_entry(build_tool_catalogue(TENANT_ID), SERVER, ACTION)
+    assert entry is not None
+    assert entry["output_schema"] == {}
+    ref = next(item for item in resources.list_tenant_resources(TENANT_ID).tools if item.id == f"{SERVER}/{ACTION}")
+    assert ref.output_schema_state == "absent"
+    assert bound_graph([entry])["nodes"][1]["data"]["output_schema"] == {}
+
+
+def test_mcp_destructive_evidence_remains_unconfirmable() -> None:
+    graph = {
+        "nodes": [
+            {
+                "id": "delete",
+                "data": {
+                    "type": "tool",
+                    "provider_type": "mcp",
+                    "provider_id": SERVER,
+                    "tool_name": "delete_documents",
+                },
+            }
+        ]
+    }
+    disclosure = disclose(graph)
+    assert disclosure.items[0].tier == "destructive"
+    assert disclosure.confirmation(["delete"]) == {}
+    assert disclosure.mocked_node_ids({}) == frozenset({"delete"})
+
+
+def test_malformed_mcp_action_does_not_hide_other_cached_actions(inventory: Engine) -> None:
+    add_provider(inventory, tools=[{"name": "bad", "inputSchema": {"type": "impossible"}}, definition()])
+    entries = [entry for entry in build_tool_catalogue(TENANT_ID) if entry["provider_type"] == "mcp"]
+    assert [entry["tool_name"] for entry in entries] == [ACTION]