feat: install plugin dep support pypi index url

Author: fatelei

.env.example

@@ -188,3 +188,6 @@ PLUGIN_RUNTIME_MAX_BUFFER_SIZE=5242880
 DIFY_BACKWARDS_INVOCATION_WRITE_TIMEOUT=5000
 # dify backwards invocation read timeout in milliseconds
 DIFY_BACKWARDS_INVOCATION_READ_TIMEOUT=240000
+
+PIP_MIRROR_URL=
+PIP_EXTRA_INDEX_URL=

README.md

@@ -62,6 +62,28 @@ Daemon uses `uv` to manage the dependencies of plugins, before you start the dae
 #### Interpreter
 There is a possibility that you have multiple python versions installed on your machine, a variable `PYTHON_INTERPRETER_PATH` is provided to specify the python interpreter path for you.
 
+#### Speeding up Python dependency installation (uv/pip)
+You can speed up plugin dependency installation by configuring Python package indexes via environment variables (set them in your shell or `.env`). The daemon reads these into its config on startup.
+
+Supported variables
+- PIP_MIRROR_URL: Primary index for both uv sync and uv pip install
+- PIP_EXTRA_INDEX_URL: One or more extra indexes (comma or space separated)
+
+Behavior
+- Applies uniformly to both dependency styles:
+  - pyproject.toml (using `uv sync`)
+  - requirements.txt (using `uv pip install`)
+- Trusted hosts are derived automatically from the configured URLs.
+
+Examples
+```bash
+# Use the official PyPI index
+PIP_MIRROR_URL=https://pypi.org/simple
+
+# Add multiple mirrors (comma or space separated)
+PIP_EXTRA_INDEX_URL="https://my.mirror/simple, https://another.mirror/simple"
+```
+
 ## Deployment
 
 Currently, the daemon only supports Linux and MacOS, lots of adaptions are needed for Windows, feel free to contribute if you need it.

internal/core/local_runtime/dependency_installation_test.go

@@ -200,6 +200,7 @@ func TestPreparePipArgs(t *testing.T) {
 			"pip",
 			"install",
 			"-i", "https://pypi.tuna.tsinghua.edu.cn/simple",
+			"--trusted-host", "pypi.tuna.tsinghua.edu.cn",
 			"-r", "requirements.txt",
 		}, args)
 	})
@@ -229,6 +230,7 @@ func TestPreparePipArgs(t *testing.T) {
 			"pip",
 			"install",
 			"-i", "https://pypi.tuna.tsinghua.edu.cn/simple",
+			"--trusted-host", "pypi.tuna.tsinghua.edu.cn",
 			"-r", "requirements.txt",
 			"-vvv",
 			"--no-cache",

internal/core/local_runtime/setup_python_environment.go

@@ -1,17 +1,22 @@
 package local_runtime
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"net/url"
 	"os"
 	"os/exec"
 	"path"
 	"path/filepath"
+	"regexp"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	routinepkg "github.com/langgenius/dify-plugin-daemon/pkg/routine"
@@ -37,8 +42,15 @@ func (p *LocalPluginRuntime) prepareUV() (string, error) {
 func (p *LocalPluginRuntime) preparePipArgs() []string {
 	args := []string{"install"}
 
-	if p.appConfig.PipMirrorUrl != "" {
-		args = append(args, "-i", p.appConfig.PipMirrorUrl)
+	// Determine index URL precedence for pip install:
+	indexURL := p.appConfig.PipMirrorUrl
+	// Extra index URLs (comma or space separated); fallback to UV extras
+	extra := p.appConfig.PipExtraIndexUrl
+	args = addIndexArgs(args, indexURL, extra)
+
+	// Derive trusted-host from index/extra URLs
+	for _, h := range deriveTrustedHosts(indexURL, extra) {
+		args = append(args, "--trusted-host", h)
 	}
 
 	args = append(args, "-r", "requirements.txt")
@@ -60,9 +72,11 @@ func (p *LocalPluginRuntime) preparePipArgs() []string {
 func (p *LocalPluginRuntime) prepareSyncArgs() []string {
 	args := []string{"sync", "--no-dev"}
 
-	if p.appConfig.PipMirrorUrl != "" {
-		args = append(args, "-i", p.appConfig.PipMirrorUrl)
-	}
+	// Determine index URL precedence for uv sync:
+	indexURL := p.appConfig.PipMirrorUrl
+	// Extra index URLs; fallback to pip extras
+	extra := p.appConfig.PipExtraIndexUrl
+	args = addIndexArgs(args, indexURL, extra)
 
 	if p.appConfig.PipVerbose {
 		args = append(args, "-v")
@@ -113,6 +127,15 @@ func (p *LocalPluginRuntime) installDependencies(
 	virtualEnvPath := path.Join(p.State.WorkingPath, ".venv")
 	cmd := exec.CommandContext(ctx, uvPath, args...)
 	cmd.Env = append(cmd.Env, "VIRTUAL_ENV="+virtualEnvPath, "PATH="+os.Getenv("PATH"))
+
+	// Also provide PIP_TRUSTED_HOST env (space-separated) for pip under uv
+	pipIndex := p.appConfig.PipMirrorUrl
+	pipExtra := p.appConfig.PipExtraIndexUrl
+
+	if hosts := deriveTrustedHosts(pipIndex, pipExtra); len(hosts) > 0 {
+		cmd.Env = append(cmd.Env, fmt.Sprintf("PIP_TRUSTED_HOST=%s", strings.Join(hosts, " ")))
+	}
+
 	if p.appConfig.HttpProxy != "" {
 		cmd.Env = append(cmd.Env, fmt.Sprintf("HTTP_PROXY=%s", p.appConfig.HttpProxy))
 	}
@@ -124,6 +147,20 @@ func (p *LocalPluginRuntime) installDependencies(
 	}
 	cmd.Dir = p.State.WorkingPath
 
+	// log start with sanitized args
+	startAt := time.Now()
+	sanitized := sanitizeArgs(args)
+	log.Info("starting dependency installation",
+		"plugin", p.Config.Identity(),
+		"method", func() string {
+			if dependencyFileType == pyprojectTomlFile {
+				return "uv sync"
+			} else {
+				return "uv pip install"
+			}
+		}(),
+		"args", strings.Join(sanitized, " "))
+
 	// get stdout and stderr
 	stdout, err := cmd.StdoutPipe()
 	if err != nil {
@@ -149,26 +186,34 @@ func (p *LocalPluginRuntime) installDependencies(
 	}()
 
 	var errMsg strings.Builder
+	var errMu sync.Mutex
 	var wg sync.WaitGroup
 	wg.Add(2)
 
-	lastActiveAt := time.Now()
+	var lastActiveAt atomic.Int64
+	lastActiveAt.Store(time.Now().UnixNano())
 
 	routine.Submit(routinepkg.Labels{
 		routinepkg.RoutineLabelKeyModule: "plugin_manager",
 		routinepkg.RoutineLabelKeyMethod: "InitPythonEnvironment",
 	}, func() {
 		defer wg.Done()
-		// read stdout
-		buf := make([]byte, 1024)
-		for {
-			n, err := stdout.Read(buf)
-			if err != nil {
-				break
-			}
-			// FIXME: move the log to separated layer
-			log.Info("installing plugin", "plugin", p.Config.Identity(), "output", string(buf[:n]))
-			lastActiveAt = time.Now()
+		// read stdout line by line
+		scanner := bufio.NewScanner(stdout)
+		buf := make([]byte, 0, 64*1024)
+		scanner.Buffer(buf, 10*1024*1024)
+		for scanner.Scan() {
+			line := scanner.Text()
+			log.Info("install deps", "plugin", p.Config.Identity(), "stream", "stdout", "line", line)
+			lastActiveAt.Store(time.Now().UnixNano())
+		}
+		if err := scanner.Err(); err != nil {
+			errMu.Lock()
+			errMsg.WriteString("stdout scan error: ")
+			errMsg.WriteString(err.Error())
+			errMsg.WriteString("\n")
+			errMu.Unlock()
+			log.Warn("install deps", "plugin", p.Config.Identity(), "stream", "stdout", "scanner_err", err.Error())
 		}
 	})
 
@@ -177,22 +222,26 @@ func (p *LocalPluginRuntime) installDependencies(
 		routinepkg.RoutineLabelKeyMethod: "InitPythonEnvironment",
 	}, func() {
 		defer wg.Done()
-		// read stderr
-		buf := make([]byte, 1024)
-		for {
-			n, err := stderr.Read(buf)
-			if err != nil && err != os.ErrClosed {
-				lastActiveAt = time.Now()
-				errMsg.WriteString(string(buf[:n]))
-				break
-			} else if err == os.ErrClosed {
-				break
-			}
-
-			if n > 0 {
-				errMsg.WriteString(string(buf[:n]))
-				lastActiveAt = time.Now()
-			}
+		// read stderr line by line
+		scanner := bufio.NewScanner(stderr)
+		buf := make([]byte, 0, 64*1024)
+		scanner.Buffer(buf, 10*1024*1024)
+		for scanner.Scan() {
+			line := scanner.Text()
+			errMu.Lock()
+			errMsg.WriteString(line)
+			errMsg.WriteString("\n")
+			errMu.Unlock()
+			log.Warn("install deps", "plugin", p.Config.Identity(), "stream", "stderr", "line", line)
+			lastActiveAt.Store(time.Now().UnixNano())
+		}
+		if err := scanner.Err(); err != nil {
+			errMu.Lock()
+			errMsg.WriteString("stderr scan error: ")
+			errMsg.WriteString(err.Error())
+			errMsg.WriteString("\n")
+			errMu.Unlock()
+			log.Warn("install deps", "plugin", p.Config.Identity(), "stream", "stderr", "scanner_err", err.Error())
 		}
 	})
 
@@ -207,14 +256,16 @@ func (p *LocalPluginRuntime) installDependencies(
 				break
 			}
 
-			if time.Since(lastActiveAt) > time.Duration(
+			if time.Since(time.Unix(0, lastActiveAt.Load())) > time.Duration(
 				p.appConfig.PythonEnvInitTimeout,
 			)*time.Second {
 				cmd.Process.Kill()
+				errMu.Lock()
 				errMsg.WriteString(fmt.Sprintf(
 					"init process exited due to no activity for %d seconds",
 					p.appConfig.PythonEnvInitTimeout,
 				))
+				errMu.Unlock()
 				break
 			}
 		}
@@ -223,12 +274,29 @@ func (p *LocalPluginRuntime) installDependencies(
 	wg.Wait()
 
 	if err := cmd.Wait(); err != nil {
+		log.Error("dependency installation failed", "plugin", p.Config.Identity(), "duration", time.Since(startAt).String(), "error", err)
 		return fmt.Errorf("failed to install dependencies: %s, output: %s", err, errMsg.String())
 	}
 
+	log.Info("dependency installation finished", "plugin", p.Config.Identity(), "duration", time.Since(startAt).String())
 	return nil
 }
 
+// sanitizeArgs redacts credentials in any URL-like arguments to avoid leaking secrets in logs.
+func sanitizeArgs(args []string) []string {
+	// Match https://user:pass@ and https://user@
+	reWithPass := regexp.MustCompile(`(https?://)[^/@:]+:[^/@]+@`)
+	reUserOnly := regexp.MustCompile(`(https?://)[^/@:]+@`)
+
+	out := make([]string, len(args))
+	for i, a := range args {
+		s := reWithPass.ReplaceAllString(a, "${1}****:****@")
+		s = reUserOnly.ReplaceAllString(s, "${1}****:****@")
+		out[i] = s
+	}
+	return out
+}
+
 type PythonVirtualEnvironment struct {
 	pythonInterpreterPath string
 }
@@ -356,6 +424,69 @@ func (p *LocalPluginRuntime) markVirtualEnvironmentAsValid() error {
 	return nil
 }
 
+// splitByCommaOrSpace splits a list like "a,b c" into tokens.
+func splitByCommaOrSpace(s string) []string {
+	// replace comma with space then split by spaces
+	s = strings.ReplaceAll(s, ",", " ")
+	fields := strings.Fields(s)
+	return fields
+}
+
+// selectURL returns the first non-empty URL from the provided list.
+func selectURL(urls ...string) string {
+	for _, u := range urls {
+		if u != "" {
+			return u
+		}
+	}
+	return ""
+}
+
+// addIndexArgs appends index and extra-index URL arguments to args.
+func addIndexArgs(args []string, indexURL string, extraIndexURL string) []string {
+	if indexURL != "" {
+		args = append(args, "-i", indexURL)
+	}
+	if extraIndexURL != "" {
+		for _, u := range splitByCommaOrSpace(extraIndexURL) {
+			if u != "" {
+				args = append(args, "--extra-index-url", u)
+			}
+		}
+	}
+	return args
+}
+
+// deriveTrustedHosts parses hostnames from index/extra URLs and returns a de-duplicated list.
+func deriveTrustedHosts(indexURL string, extraIndexURL string) []string {
+	set := map[string]struct{}{}
+	add := func(raw string) {
+		if strings.TrimSpace(raw) == "" {
+			return
+		}
+		u, err := url.Parse(raw)
+		if err != nil || u.Host == "" {
+			return
+		}
+		host := u.Host
+		if i := strings.Index(host, ":"); i >= 0 {
+			host = host[:i]
+		}
+		set[host] = struct{}{}
+	}
+	add(indexURL)
+	for _, raw := range splitByCommaOrSpace(extraIndexURL) {
+		add(raw)
+	}
+	out := make([]string, 0, len(set))
+	for h := range set {
+		out = append(out, h)
+	}
+	// preserve deterministic order: sort hostnames
+	sort.Strings(out)
+	return out
+}
+
 func (p *LocalPluginRuntime) preCompile(
 	pythonPath string,
 ) error {