Dify 的插件不是"在主进程里 import 一段代码"，而是跑在一个独立的 plugin daemon（守护进程/服务）里。主程序（api/）通过 HTTP 内部 API 和它对话。配置里能看到这个 daemon 的地址（core/plugin/impl/base.py:43）：

# api/core/plugin/impl/base.py:43
plugin_daemon_inner_api_baseurl = URL(str(dify_config.PLUGIN_DAEMON_URL))   # ★ daemon 的地址
...
PLUGIN_DAEMON_MAX_PATH_LENGTH = 4096          # 路径长度上限（安全）
PLUGIN_DAEMON_MAX_PATH_DECODE_DEPTH = 8       # 路径解码深度上限（防编码攻击）

所有和 daemon 的通信都走一个基类客户端 BasePluginClient（core/plugin/impl/base.py:68），核心是 _request（impl/base.py:69）：

# api/core/plugin/impl/base.py:69
def _request(self, method, path, headers=None, data=None, params=None, files=None) -> httpx.Response:
    """Make a request to the plugin daemon inner API."""
    url, headers, prepared_data, params, files = self._prepare_request(path, headers, data, params, files)
    request_kwargs = {"method": method, "url": url, "headers": headers,
                      "params": params, "files": files, "timeout": plugin_daemon_request_timeout}
    ...
    try:
        response = _httpx_client.request(**request_kwargs)          # ★ 就是发一个 HTTP 请求给 daemon
    except httpx.RequestError:
        logger.exception("Request to Plugin Daemon Service failed")
        raise PluginDaemonInnerError(code=-500, message="Request to Plugin Daemon Service failed")
    return response

请求前 _prepare_request 做了两件关键的安全 + 鉴权处理（impl/base.py:128）：

# api/core/plugin/impl/base.py (_prepare_request 节选)
if any(seg == ".." for seg in decoded_path.split("/")):             # ① 防路径穿越
    raise ValueError(f"Invalid plugin daemon path: traversal sequence detected in {path!r}")
url = plugin_daemon_inner_api_baseurl / path
prepared_headers = dict(headers or {})
prepared_headers["X-Api-Key"] = dify_config.PLUGIN_DAEMON_KEY       # ★② 用内部密钥鉴权
self._inject_trace_headers(prepared_headers)                        # ③ 注入分布式追踪头（Day19）

面向业务的门面是 PluginService（core/plugin/plugin_service.py:81）。它不自己发 HTTP，而是委托给各种 Installer/Manager（它们底层都是 L03 的 BasePluginClient）。比如"列出某租户装了哪些插件"：

# api/core/plugin/plugin_service.py (list_plugins 节选)
manager = PluginInstaller()
plugins = manager.list_plugins(tenant_id)     # ★ 委托给 installer，底层向 daemon 发请求
...
def fetch_plugin_model_providers(...):        # 拉某插件提供了哪些模型供应商
    ...

MCP（Model Context Protocol）是一套开放协议，用 JSON-RPC 规范请求/响应。Dify 在 core/mcp/types.py 里声明了它遵循的协议版本和消息格式（core/mcp/types.py:26）：

# api/core/mcp/types.py:26
LATEST_PROTOCOL_VERSION = "2025-06-18"          # ★ Dify 支持的 MCP 协议版本
SERVER_LATEST_PROTOCOL_VERSION = "2025-06-18"
...
# api/core/mcp/types.py:178
class JSONRPCMessage(RootModel[JSONRPCRequest | JSONRPCNotification
                             | JSONRPCResponse | JSONRPCError]):    # ★ 一条 MCP 消息 = 四种之一
    ...

MCPClient（core/mcp/mcp_client.py:20）是 Dify 作为 MCP 客户端的门面。它连接时有个漂亮的"自动降级"设计（mcp_client.py:61）：

# api/core/mcp/mcp_client.py (connect 节选)
methods = {
    "mcp": streamablehttp_client,      # 首选：streamable http（较新的传输方式）
    "sse":  sse_client,                # 备选：SSE
}
...
    self.connect_server(sse_client, "sse")            # 先试 SSE
except ...:
    logger.debug("MCP connection failed with 'sse', falling back to 'mcp' method.")
    self.connect_server(streamablehttp_client, "mcp") # ★ 失败就降级换另一种传输

连上以后，"列工具"和"调工具"只是两行——它们把请求转成 L05 的 JSON-RPC 消息发给服务器（mcp_client.py:107 与 mcp_client.py:114）：

# api/core/mcp/mcp_client.py:107
def list_tools(self) -> list[Tool]:
    if not self._session:
        raise ValueError("Session not initialized.")
    response = self._session.list_tools()             # 问服务器"你有哪些工具"
    return response.tools

# api/core/mcp/mcp_client.py:114
def invoke_tool(self, tool_name: str, tool_args: dict[str, Any]) -> CallToolResult:
    if not self._session:
        raise ValueError("Session not initialized.")
    return self._session.call_tool(tool_name, tool_args)   # ★ 调用某个工具（发 tools/call 请求）