Method validate_mcp_transport

apps/common/utils/tool_code.py:381–385

(self, code_str)

            raise Exception(_(f"Process execution timed out after {_process_limit_timeout_seconds} seconds."))

    def validate_mcp_transport(self, code_str):
        servers = json.loads(code_str)
        for server, config in servers.items():
            if config.get("transport") not in ["sse", "streamable_http"]:
                raise Exception(_("Only support transport=sse or transport=streamable_http"))


@contextmanager