refactor: replace MultiServerMCPClient with create_mcp_client and update MCP server validation

apps/application/chat_pipeline/step/chat_step/impl/base_chat_step.py

@@ -445,7 +445,7 @@ def _handle_mcp_request(
                 chat_model,
                 system_prompt,
                 message_list,
-                json.dumps(mcp_servers_config),
+                mcp_servers_config,
                 mcp_output_enable,
                 tool_init_params,
                 source_id,

apps/application/flow/step_node/ai_chat_step_node/impl/base_chat_node.py

@@ -446,7 +446,7 @@ def _handle_mcp_request(
                 chat_model,
                 system_prompt,
                 message_list,
-                json.dumps(mcp_servers_config),
+                mcp_servers_config,
                 mcp_output_enable,
                 tool_init_params,
                 source_id,

apps/application/flow/step_node/mcp_node/impl/base_mcp_node.py

@@ -4,7 +4,7 @@
 from typing import List
 
 from django.db.models import QuerySet
-from langchain_mcp_adapters.client import MultiServerMCPClient
+from common.utils.mcp_client import create_mcp_client
 
 from application.flow.i_step_node import NodeResult
 from application.flow.step_node.mcp_node.i_mcp_node import IMcpNode
@@ -46,7 +46,7 @@ def execute(self, mcp_servers, mcp_server, mcp_tool, mcp_tool_id, mcp_source, to
         params = self.handle_variables(params)
 
         async def call_tool(t, a):
-            client = MultiServerMCPClient(servers)
+            client = create_mcp_client(servers)
             async with client.session(mcp_server) as s:
                 return await s.call_tool(t, a)
 

apps/application/flow/tools.py

@@ -45,7 +45,7 @@
 from langchain_core.messages import AIMessageChunk, BaseMessage, BaseMessageChunk, ToolMessage
 from langchain_core.tools import StructuredTool
 from langchain_core.utils._merge import merge_lists as _original_merge_lists
-from langchain_mcp_adapters.client import MultiServerMCPClient
+from common.utils.mcp_client import create_mcp_client
 from langgraph.checkpoint.memory import MemorySaver
 from maxkb.const import CONFIG
 from pydantic import Field, create_model
@@ -396,7 +396,7 @@ def _extract_tool_id(raw_id):
 
 async def _initialize_skills(mcp_servers, temp_dir):
     skills_dir = os.path.join(temp_dir, "skills")
-    mcp_config = json.loads(mcp_servers)
+    mcp_config = dict(mcp_servers)  # Preserve server-generated InternalMCPConfig objects.
     if "skills" in mcp_config:
         skill_file_items = mcp_config.pop("skills")
         for skill_file in skill_file_items:
@@ -435,7 +435,7 @@ async def _initialize_skills(mcp_servers, temp_dir):
 
         os.system("chmod -R g+rx " + temp_dir)  # 确保技能目录可访问
 
-    client = MultiServerMCPClient(mcp_config)
+    client = create_mcp_client(mcp_config)
 
     return client
 

apps/application/serializers/application.py

@@ -51,7 +51,7 @@
 from knowledge.models import File, FileSourceType, Knowledge, KnowledgeScope
 from knowledge.serializers.common import BatchMoveSerializer, BatchSerializer
 from knowledge.serializers.knowledge import KnowledgeModelSerializer, KnowledgeSerializer
-from langchain_mcp_adapters.client import MultiServerMCPClient
+from common.utils.mcp_client import create_mcp_client
 from maxkb.conf import PROJECT_DIR
 from maxkb.const import CONFIG
 from models_provider.base_model_provider import ModelTypeConst
@@ -1154,7 +1154,7 @@ class PlayDemoTextRequest(serializers.Serializer):
 
 
 async def get_mcp_tools(servers):
-    client = MultiServerMCPClient(servers)
+    client = create_mcp_client(servers)
     return await client.get_tools()
 
 
@@ -1181,9 +1181,7 @@ def get_mcp_servers(self, instance, with_valid=True):
             self.is_valid(raise_exception=True)
             McpServersSerializer(data=instance).is_valid(raise_exception=True)
         servers = json.loads(instance.get("mcp_servers"))
-        for server, config in servers.items():
-            if config.get("transport") not in ["sse", "streamable_http"]:
-                raise AppApiException(500, _("Only support transport=sse or transport=streamable_http"))
+        ToolExecutor().validate_mcp_transport(json.dumps(servers))
         tools = []
         for server in servers:
             tools += [

apps/common/utils/mcp_client.py

@@ -0,0 +1,72 @@
+"""MCP egress policy shared by validation, discovery and tool execution."""
+
+import ipaddress
+from functools import partial
+
+from langchain_mcp_adapters.client import MultiServerMCPClient
+
+from common.utils.mcp_network import check_addresses, http_client_factory, parse_url
+
+
+class InternalMCPConfig(dict):
+    """In-memory provenance for configurations generated by ToolExecutor.
+
+    Never deserialize user input into this type. JSON round trips deliberately
+    lose this privilege; keep runtime configurations in memory instead.
+    """
+
+
+def allowed_networks():
+    from maxkb.const import CONFIG
+
+    return tuple(
+        ipaddress.ip_network(value.strip())
+        for value in CONFIG.get("MCP_ALLOWED_NETWORKS", "").split(",")
+        if value.strip()
+    )
+
+
+def validate_mcp_servers(servers):
+    if not isinstance(servers, dict):
+        raise ValueError("MCP servers must be an object")
+    networks = allowed_networks()
+    for config in servers.values():
+        if not isinstance(config, dict) or config.get("transport") not in ("sse", "streamable_http"):
+            raise ValueError("Only support transport=sse or transport=streamable_http")
+        url = parse_url(config.get("url"))
+        # Do not resolve user hostnames in the web process. The worker checks
+        # every DNS result at connection time, including nonstandard IP notation.
+        try:
+            address = ipaddress.ip_address(url.host)
+        except ValueError:
+            continue
+        check_addresses([str(address)], networks)
+
+
+def create_mcp_client(servers):
+    if not isinstance(servers, dict):
+        raise ValueError("MCP servers must be an object")
+    connections = {}
+    networks = allowed_networks()
+    for name, config in servers.items():
+        if not isinstance(config, dict):
+            raise ValueError("MCP server configuration must be an object")
+        internal = isinstance(config, InternalMCPConfig)
+        if internal and config.get("transport") == "stdio":
+            connections[name] = dict(config)
+            continue
+        if config.get("transport") not in ("sse", "streamable_http"):
+            raise ValueError("Only support transport=sse or transport=streamable_http")
+        url = parse_url(config.get("url"))
+        if internal:
+            connection = dict(config)
+            connection["httpx_client_factory"] = partial(
+                http_client_factory, url=str(url), internal=True,
+                networks=(ipaddress.ip_network("127.0.0.1/32"),),
+            )
+        else:
+            from common.utils.mcp_sandbox import sandbox_connection
+
+            connection = sandbox_connection(config, networks)
+        connections[name] = connection
+    return MultiServerMCPClient(connections)

apps/common/utils/mcp_network.py

@@ -0,0 +1,138 @@
+"""HTTP destination checks usable without Django inside an MCP worker."""
+
+import ipaddress
+import socket
+import ssl
+
+import anyio
+import httpcore
+import httpx
+
+
+class MCPNetworkPolicyError(ValueError):
+    """Locally generated, credential-free policy failure safe to report."""
+
+
+def sandbox_failure_message(error):
+    # SDK exception groups and chained HTTP errors can embed credentials. Only
+    # report our own policy text, a numeric HTTP status, or a fixed description.
+    errors, pending, seen = [], [error], set()
+    while pending:
+        current = pending.pop()
+        if id(current) in seen:
+            continue
+        seen.add(id(current))
+        errors.append(current)
+        if isinstance(current, BaseExceptionGroup):
+            pending.extend(current.exceptions)
+        if current.__cause__ is not None:
+            pending.append(current.__cause__)
+    for current in errors:
+        if isinstance(current, MCPNetworkPolicyError):
+            return str(current)
+    for current in errors:
+        if isinstance(current, httpx.HTTPStatusError):
+            status = current.response.status_code
+            if 300 <= status < 400:
+                return "MCP endpoint returned a redirect; configure its final URL"
+            return f"MCP endpoint returned HTTP {status}; check endpoint and credentials"
+    for exception_type, message in (
+        (ssl.SSLCertVerificationError, "MCP TLS certificate verification failed"),
+        (socket.gaierror, "MCP hostname resolution failed; check container DNS"),
+        (PermissionError, "MCP access denied; check sandbox file and network policy"),
+        ((httpx.TimeoutException, TimeoutError), "MCP connection timed out"),
+        (httpx.ConnectError, "MCP connection failed; check container connectivity and sandbox network policy"),
+    ):
+        if any(isinstance(current, exception_type) for current in errors):
+            return message
+    return "MCP session failed; check endpoint, sandbox setup and network policy"
+
+
+def parse_url(value):
+    if not isinstance(value, str) or not value or any(ord(c) <= 32 for c in value):
+        raise ValueError("Invalid MCP server URL")
+    try:
+        url = httpx.URL(value)
+        if (
+            url.scheme not in ("http", "https") or not url.host or url.userinfo
+            or url.fragment or "%" in url.host or "\\" in value
+            or (url.port is not None and not 1 <= url.port <= 65535)
+        ):
+            raise ValueError("Invalid MCP server URL")
+        return url
+    except (httpx.InvalidURL, ValueError) as exc:
+        raise ValueError("Invalid MCP server URL") from exc
+
+
+def check_addresses(addresses, networks):
+    if not addresses:
+        raise MCPNetworkPolicyError("MCP server hostname has no addresses")
+    for address in addresses:
+        ip = ipaddress.ip_address(address)
+        if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped:
+            ip = ip.ipv4_mapped
+        # Do not let IPv6 transition mechanisms tunnel to restricted IPv4 hosts.
+        transition = isinstance(ip, ipaddress.IPv6Address) and (
+            ip.sixtofour is not None or ip.teredo is not None
+            or ip in ipaddress.ip_network("64:ff9b::/96")
+            or ip in ipaddress.ip_network("64:ff9b:1::/48")
+        )
+        public = ip.is_global and not ip.is_multicast and not transition
+        if not public and not any(ip in network for network in networks):
+            raise MCPNetworkPolicyError("MCP server address is not allowed by the network policy")
+
+
+class MCPNetworkBackend(httpcore.AnyIOBackend):
+    def __init__(self, networks):
+        self.networks = networks
+
+    async def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None):
+        try:
+            with anyio.fail_after(timeout):
+                # Resolve again at connection time, validate EVERY result, then
+                # connect to the numeric address. HTTP Host and TLS SNI remain
+                # the original hostname in httpcore, including certificate checks.
+                results = await anyio.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+                addresses = list(dict.fromkeys(item[4][0] for item in results))
+                check_addresses(addresses, self.networks)
+                for index, address in enumerate(addresses):
+                    try:
+                        return await super().connect_tcp(
+                            address, port, timeout, local_address, socket_options
+                        )
+                    except (httpcore.ConnectError, httpcore.ConnectTimeout):
+                        if index == len(addresses) - 1:
+                            raise
+        except TimeoutError as exc:
+            raise httpcore.ConnectTimeout() from exc
+        except OSError as exc:
+            raise httpcore.ConnectError(str(exc)) from exc
+
+
+class MCPTransport(httpx.AsyncHTTPTransport):
+    def __init__(self, url, networks, internal=False):
+        super().__init__(trust_env=False)
+        self.url = parse_url(url)
+        self.internal = internal
+        # HTTPX 0.28 has no public network_backend argument. Keep its standard
+        # response/error handling and replace only the pool's connection backend.
+        self._pool._network_backend = MCPNetworkBackend(networks)
+
+    async def handle_async_request(self, request):
+        target = parse_url(str(request.url))
+        if (target.scheme, target.host, target.port) != (self.url.scheme, self.url.host, self.url.port):
+            raise ValueError("MCP requests must stay on the configured origin")
+        if self.internal and target != self.url:
+            raise ValueError("Internal MCP requests must use the generated endpoint")
+        return await super().handle_async_request(request)
+
+
+def http_client_factory(headers=None, timeout=None, auth=None, *, url, networks, internal=False):
+    return httpx.AsyncClient(
+        headers=headers,
+        timeout=timeout if timeout is not None else httpx.Timeout(30, read=300),
+        auth=auth,
+        follow_redirects=False,
+        trust_env=False,
+        transport=MCPTransport(url, networks, internal),
+    )

apps/common/utils/mcp_sandbox.py

@@ -0,0 +1,61 @@
+"""Build fixed stdio worker connections; user configuration never selects code."""
+
+import json
+import pwd
+import sys
+from datetime import timedelta
+from pathlib import Path
+
+from mcp.types import Implementation
+
+
+BOOTSTRAP_KEY = "maxkbSandbox"
+REMOTE_FIELDS = {"transport", "url", "headers", "timeout", "sse_read_timeout", "terminate_on_close"}
+
+
+def sandbox_settings():
+    from maxkb.const import CONFIG
+
+    if not sys.platform.startswith("linux") or not bool(int(CONFIG.get("SANDBOX", 1))):
+        raise ValueError("External MCP requires an enabled Linux sandbox")
+    account = pwd.getpwnam("sandbox")
+    sandbox_home = Path(CONFIG.get("SANDBOX_HOME", "/opt/maxkb-app/sandbox"))
+    library = sandbox_home / "lib/sandbox.so"
+    if not library.is_file() or not library.with_name(".sandbox.conf").is_file():
+        raise ValueError("MCP sandbox library or configuration is missing")
+    return {
+        "uid": account.pw_uid,
+        "gid": account.pw_gid,
+        "library": str(library),
+        "cwd": str(sandbox_home),
+        "python_paths": CONFIG.get_sandbox_python_package_paths().split(","),
+        "memory_mb": int(CONFIG.get("SANDBOX_PYTHON_PROCESS_LIMIT_MEM_MB", "256")),
+        "cpu_cores": int(CONFIG.get("SANDBOX_PYTHON_PROCESS_LIMIT_CPU_CORES", "1")),
+        "timeout": int(CONFIG.get("SANDBOX_PYTHON_PROCESS_LIMIT_TIMEOUT_SECONDS", "3600")),
+    }
+
+
+def sandbox_connection(config, networks):
+    settings = sandbox_settings()
+    # Only transport data goes to the remote client. In particular, ignore user
+    # command/env/factory/session_kwargs fields and never deserialize Python code.
+    remote = {key: value for key, value in config.items() if key in REMOTE_FIELDS}
+    bootstrap = {"connection": remote, "networks": [str(network) for network in networks]}
+    # Check serializability before launching, and detach from mutable input.
+    bootstrap = json.loads(json.dumps(bootstrap, allow_nan=False))
+    return {
+        "transport": "stdio",
+        "command": sys.executable,
+        "args": ["-I", str(Path(__file__).with_name("mcp_sandbox_worker.py"))],
+        "cwd": settings["cwd"],
+        "env": {
+            "LD_PRELOAD": settings["library"],
+            "MAXKB_MCP_WORKER_SETTINGS": json.dumps(settings),
+        },
+        "session_kwargs": {
+            "read_timeout_seconds": timedelta(seconds=settings["timeout"]),
+            # This field travels only over the child's stdio pipe. The worker
+            # removes it before forwarding initialize to the remote server.
+            "client_info": Implementation(name="maxkb-sandbox", version="1", **{BOOTSTRAP_KEY: bootstrap}),
+        },
+    }