zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-28 19:58:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Feat: add local & ssh provider in admin panel (#15039)

Browse Source 
### What problem does this PR solve?

Feat: add local & ssh provider in admin panel

### Type of change

- [x] New Feature (non-breaking change which adds functionality)
This commit is contained in:
Magicbook1108
2026-05-20 16:56:20 +08:00
committed by GitHub
parent 6499bce2a6
commit b28e134944
19 changed files with 1836 additions and 408 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
agent/sandbox/client.py
View File

@@ -23,7 +23,6 @@ with the configured sandbox provider.
import json
import logging
import os
from typing import Dict, Any, Optional
from api.db.services.system_settings_service import SystemSettingsService
@@ -49,7 +48,6 @@ def get_provider_manager() -> ProviderManager:
if _provider_manager is not None:
return _provider_manager
# Initialize provider manager with system settings
_provider_manager = ProviderManager()
_load_provider_from_settings()
@@ -61,7 +59,7 @@ def _load_provider_from_settings() -> None:
Load sandbox provider from system settings and configure the provider manager.
This function resolves the active provider type, then loads configuration
from system settings with environment overrides for that provider.
from system settings.
"""
global _provider_manager
@@ -69,7 +67,7 @@ def _load_provider_from_settings() -> None:
return
try:
provider_type, provider_type_from_env = _resolve_provider_type()
provider_type = _resolve_provider_type()
config = _load_provider_config(provider_type)
# Import and instantiate the provider
@@ -78,6 +76,7 @@ def _load_provider_from_settings() -> None:
AliyunCodeInterpreterProvider,
E2BProvider,
LocalProvider,
SSHProvider,
)
provider_classes = {
@@ -85,11 +84,10 @@ def _load_provider_from_settings() -> None:
"aliyun_codeinterpreter": AliyunCodeInterpreterProvider,
"e2b": E2BProvider,
"local": LocalProvider,
"ssh": SSHProvider,
}
if provider_type not in provider_classes:
if provider_type_from_env:
raise SandboxProviderConfigError(f"Unknown sandbox provider type: {provider_type}")
logger.error(f"Unknown provider type: {provider_type}")
return
@@ -99,7 +97,7 @@ def _load_provider_from_settings() -> None:
# Initialize the provider
if not provider.initialize(config):
message = f"Failed to initialize sandbox provider: {provider_type}. Config keys: {list(config.keys())}"
if provider_type == "local" or provider_type_from_env:
if provider_type in {"local", "ssh"}:
raise SandboxProviderConfigError(message)
logger.error(message)
return
@@ -114,8 +112,6 @@ def _load_provider_from_settings() -> None:
logger.error(f"Failed to load sandbox provider from settings: {e}")
import traceback
traceback.print_exc()
def _load_provider_config_from_settings(provider_type: str) -> Dict[str, Any]:
provider_config_settings = SystemSettingsService.get_by_name(f"sandbox.{provider_type}")
if not provider_config_settings:
@@ -129,64 +125,15 @@ def _load_provider_config_from_settings(provider_type: str) -> Dict[str, Any]:
return {}
def _resolve_provider_type() -> tuple[str, bool]:
provider_type = os.environ.get("SANDBOX_PROVIDER_TYPE", "").strip()
if provider_type:
return provider_type, True
def _resolve_provider_type() -> str:
provider_type_settings = SystemSettingsService.get_by_name("sandbox.provider_type")
if not provider_type_settings:
raise RuntimeError(
"Sandbox provider type not configured. Please set 'sandbox.provider_type' in system settings."
)
return provider_type_settings[0].value, False
return "self_managed"
return provider_type_settings[0].value
def _load_provider_config(provider_type: str) -> Dict[str, Any]:
config = _load_provider_config_from_settings(provider_type)
env_config = _load_provider_config_from_env(provider_type)
if env_config:
config.update(env_config)
return config
def _load_provider_config_from_env(provider_type: str) -> Dict[str, Any]:
if provider_type == "local":
return _load_local_provider_config_from_env()
if provider_type == "self_managed":
return _load_self_managed_provider_config_from_env()
return {}
def _load_local_provider_config_from_env() -> Dict[str, Any]:
env_to_config = {
"SANDBOX_LOCAL_PYTHON_BIN": "python_bin",
"SANDBOX_LOCAL_NODE_BIN": "node_bin",
"SANDBOX_LOCAL_WORK_DIR": "work_dir",
"SANDBOX_LOCAL_TIMEOUT": "timeout",
"SANDBOX_LOCAL_MAX_MEMORY_MB": "max_memory_mb",
"SANDBOX_LOCAL_MAX_OUTPUT_BYTES": "max_output_bytes",
"SANDBOX_LOCAL_MAX_ARTIFACTS": "max_artifacts",
"SANDBOX_LOCAL_MAX_ARTIFACT_BYTES": "max_artifact_bytes",
}
config = {}
for env_name, config_name in env_to_config.items():
if env_name in os.environ:
config[config_name] = os.environ[env_name]
return config
def _load_self_managed_provider_config_from_env() -> Dict[str, Any]:
host = os.environ.get("SANDBOX_HOST", "").strip()
port = os.environ.get("SANDBOX_EXECUTOR_MANAGER_PORT", "").strip()
pool_size = os.environ.get("SANDBOX_EXECUTOR_MANAGER_POOL_SIZE", "").strip()
config = {}
if host:
config["endpoint"] = f"http://{host}:{port or '9385'}"
if pool_size:
config["pool_size"] = pool_size
return config
return _load_provider_config_from_settings(provider_type)
def reload_provider() -> None:
@@ -231,6 +178,14 @@ def execute_code(
)
provider = provider_manager.get_provider()
provider_name = provider_manager.get_provider_name() or getattr(provider, "__class__", type(provider)).__name__
logger.info(
"CodeExec using sandbox provider '%s' (language=%s, timeout=%ss)",
provider_name,
language,
timeout,
)
# Create a sandbox instance
instance = provider.create_instance(template=language)

3
agent/sandbox/providers/__init__.py
View File

@@ -25,6 +25,7 @@ This package contains:
Official Documentation: https://help.aliyun.com/zh/functioncompute/fc/sandbox-sandbox-code-interepreter
- e2b.py: E2B provider implementation
- local.py: Local process provider implementation
- ssh.py: Remote SSH provider implementation
"""
from .base import SandboxProvider, SandboxInstance, ExecutionResult, SandboxProviderConfigError
@@ -33,6 +34,7 @@ from .self_managed import SelfManagedProvider
from .aliyun_codeinterpreter import AliyunCodeInterpreterProvider
from .e2b import E2BProvider
from .local import LocalProvider
from .ssh import SSHProvider
__all__ = [
"SandboxProvider",
@@ -44,4 +46,5 @@ __all__ = [
"AliyunCodeInterpreterProvider",
"E2BProvider",
"LocalProvider",
"SSHProvider",
]

106
agent/sandbox/providers/local.py
View File

@@ -49,12 +49,6 @@ LOCAL_PYTHON_THREAD_ENV_VARS = (
"BLIS_NUM_THREADS",
"VECLIB_MAXIMUM_THREADS",
)
def _env_enabled(name: str) -> bool:
return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
class LocalProvider(SandboxProvider):
"""
Execute code as a local child process.
@@ -76,17 +70,14 @@ class LocalProvider(SandboxProvider):
self._instances: dict[str, Path] = {}
def initialize(self, config: Dict[str, Any]) -> bool:
if not _env_enabled("SANDBOX_LOCAL_ENABLED"):
raise SandboxProviderConfigError("Local code execution is disabled. Set SANDBOX_LOCAL_ENABLED=true to enable it.")
self.python_bin = str(self._resolve_config_value(config, "python_bin", "SANDBOX_LOCAL_PYTHON_BIN", "python3"))
self.node_bin = str(self._resolve_config_value(config, "node_bin", "SANDBOX_LOCAL_NODE_BIN", "node"))
self.work_dir = Path(self._resolve_config_value(config, "work_dir", "SANDBOX_LOCAL_WORK_DIR", "/tmp/ragflow-codeexec")).resolve()
self.timeout = int(self._resolve_config_value(config, "timeout", "SANDBOX_LOCAL_TIMEOUT", 30))
self.max_memory_mb = int(self._resolve_config_value(config, "max_memory_mb", "SANDBOX_LOCAL_MAX_MEMORY_MB", 512))
self.max_output_bytes = int(self._resolve_config_value(config, "max_output_bytes", "SANDBOX_LOCAL_MAX_OUTPUT_BYTES", 1024 * 1024))
self.max_artifacts = int(self._resolve_config_value(config, "max_artifacts", "SANDBOX_LOCAL_MAX_ARTIFACTS", 20))
self.max_artifact_bytes = int(self._resolve_config_value(config, "max_artifact_bytes", "SANDBOX_LOCAL_MAX_ARTIFACT_BYTES", 10 * 1024 * 1024))
self.python_bin = str(config.get("python_bin", "python3"))
self.node_bin = str(config.get("node_bin", "node"))
self.work_dir = Path(str(config.get("work_dir", "/tmp/ragflow-codeexec"))).resolve()
self.timeout = int(config.get("timeout", 30))
self.max_memory_mb = int(config.get("max_memory_mb", 512))
self.max_output_bytes = int(config.get("max_output_bytes", 1024 * 1024))
self.max_artifacts = int(config.get("max_artifacts", 20))
self.max_artifact_bytes = int(config.get("max_artifact_bytes", 10 * 1024 * 1024))
self._validate_limits()
self.work_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
@@ -194,14 +185,72 @@ class LocalProvider(SandboxProvider):
@staticmethod
def get_config_schema() -> Dict[str, Dict]:
return {
"python_bin": {"type": "string", "required": False, "default": "python3"},
"node_bin": {"type": "string", "required": False, "default": "node"},
"work_dir": {"type": "string", "required": False, "default": "/tmp/ragflow-codeexec"},
"timeout": {"type": "integer", "required": False, "default": 30},
"max_memory_mb": {"type": "integer", "required": False, "default": 512},
"max_output_bytes": {"type": "integer", "required": False, "default": 1048576},
"max_artifacts": {"type": "integer", "required": False, "default": 20},
"max_artifact_bytes": {"type": "integer", "required": False, "default": 10485760},
"python_bin": {
"type": "string",
"required": False,
"default": "python3",
"label": "Python Binary",
"description": "Python executable used for local code execution.",
},
"node_bin": {
"type": "string",
"required": False,
"default": "node",
"label": "Node.js Binary",
"description": "Node.js executable used for local JavaScript execution.",
},
"work_dir": {
"type": "string",
"required": False,
"default": "/tmp/ragflow-codeexec",
"label": "Working Directory",
"description": "Directory used to store temporary scripts and artifacts on the current host.",
},
"timeout": {
"type": "integer",
"required": False,
"default": 30,
"label": "Timeout (seconds)",
"description": "Maximum execution time for each local run. Unit: seconds.",
"min": 1,
"max": 600,
},
"max_memory_mb": {
"type": "integer",
"required": False,
"default": 512,
"label": "Max Memory (MB)",
"description": "Address-space memory limit for the local child process. Unit: MB.",
"min": 1,
"max": 65536,
},
"max_output_bytes": {
"type": "integer",
"required": False,
"default": 1048576,
"label": "Max Output (bytes)",
"description": "Maximum combined stdout and stderr size. Unit: bytes.",
"min": 1024,
"max": 10485760,
},
"max_artifacts": {
"type": "integer",
"required": False,
"default": 20,
"label": "Max Artifacts",
"description": "Maximum number of files collected from the artifacts directory.",
"min": 0,
"max": 100,
},
"max_artifact_bytes": {
"type": "integer",
"required": False,
"default": 10485760,
"label": "Max Artifact Size (bytes)",
"description": "Maximum size of a single artifact file. Unit: bytes.",
"min": 1024,
"max": 104857600,
},
}
def _validate_limits(self) -> None:
@@ -227,13 +276,6 @@ class LocalProvider(SandboxProvider):
return [self.node_bin, str(script_path)], script_path
raise RuntimeError(f"Unsupported language for local provider: {language}")
@staticmethod
def _resolve_config_value(config: Dict[str, Any], key: str, env_name: str, default: Any) -> Any:
value = config.get(key)
if value is not None:
return value
return os.environ.get(env_name, default)
def _build_child_env(self, instance_dir: Path) -> dict[str, str]:
env = {
"HOME": str(instance_dir),

122
agent/sandbox/providers/self_managed.py
View File

@@ -22,6 +22,7 @@ a pool of Docker containers with gVisor for secure code execution.
"""
import base64
import os
import time
import uuid
from typing import Dict, Any, List, Optional
@@ -40,10 +41,10 @@ class SelfManagedProvider(SandboxProvider):
"""
def __init__(self):
self.endpoint: str = "http://localhost:9385"
self.endpoint: str = "http://sandbox-executor-manager:9385"
self.timeout: int = 30
self.max_retries: int = 3
self.pool_size: int = 10
self.pool_size: int = 3
self._initialized: bool = False
def initialize(self, config: Dict[str, Any]) -> bool:
@@ -52,7 +53,7 @@ class SelfManagedProvider(SandboxProvider):
Args:
config: Configuration dictionary with keys:
- endpoint: HTTP endpoint (default: "http://localhost:9385")
- endpoint: HTTP endpoint (default: "http://sandbox-executor-manager:9385")
- timeout: Request timeout in seconds (default: 30)
- max_retries: Maximum retry attempts (default: 3)
- pool_size: Container pool size for info (default: 10)
@@ -60,30 +61,13 @@ class SelfManagedProvider(SandboxProvider):
Returns:
True if initialization successful, False otherwise
"""
self.endpoint = config.get("endpoint", "http://localhost:9385")
self.endpoint = config.get("endpoint", "http://sandbox-executor-manager:9385")
self.timeout = config.get("timeout", 30)
self.max_retries = config.get("max_retries", 3)
self.pool_size = config.get("pool_size", 10)
self.pool_size = config.get("executor_manager_pool_size", config.get("pool_size", 3))
# Validate endpoint is accessible
if not self.health_check():
# Try to fall back to SANDBOX_HOST from settings if we are using localhost
if "localhost" in self.endpoint or "127.0.0.1" in self.endpoint:
try:
from common import settings
if settings.SANDBOX_HOST and settings.SANDBOX_HOST not in self.endpoint:
original_endpoint = self.endpoint
self.endpoint = f"http://{settings.SANDBOX_HOST}:9385"
if self.health_check():
import logging
logging.warning(f"Sandbox self_managed: Connected using settings.SANDBOX_HOST fallback: {self.endpoint} (original: {original_endpoint})")
self._initialized = True
return True
else:
self.endpoint = original_endpoint # Restore if fallback also fails
except ImportError:
pass
return False
self._initialized = True
@@ -270,9 +254,11 @@ class SelfManagedProvider(SandboxProvider):
"type": "string",
"required": True,
"label": "Executor Manager Endpoint",
"placeholder": "http://localhost:9385",
"default": "http://localhost:9385",
"description": "HTTP endpoint of the executor_manager service"
"placeholder": "http://sandbox-executor-manager:9385",
"default": "http://sandbox-executor-manager:9385",
"description": "HTTP endpoint used by RAGFlow to call sandbox-executor-manager.",
"scope": "runtime",
"readonly": False,
},
"timeout": {
"type": "integer",
@@ -281,26 +267,86 @@ class SelfManagedProvider(SandboxProvider):
"default": 30,
"min": 5,
"max": 300,
"description": "HTTP request timeout for code execution"
"description": "Maximum request time for a single code execution call. Unit: seconds.",
"scope": "runtime",
"readonly": False,
},
"max_retries": {
"type": "integer",
"executor_manager_image": {
"type": "string",
"required": False,
"label": "Max Retries",
"default": 3,
"min": 0,
"max": 10,
"description": "Maximum number of retry attempts for failed requests"
"label": "Executor Manager Image",
"default": os.getenv("SANDBOX_EXECUTOR_MANAGER_IMAGE", "infiniflow/sandbox-executor-manager:latest"),
"description": "Docker image used by sandbox-executor-manager.",
"scope": "deployment",
"readonly": True,
},
"pool_size": {
"executor_manager_pool_size": {
"type": "integer",
"required": False,
"label": "Container Pool Size",
"default": 10,
"default": int(os.getenv("SANDBOX_EXECUTOR_MANAGER_POOL_SIZE", "3")),
"min": 1,
"max": 100,
"description": "Size of the container pool (configured in executor_manager)"
}
"description": "Container pool size used by sandbox-executor-manager.",
"scope": "deployment",
"readonly": True,
},
"base_python_image": {
"type": "string",
"required": False,
"label": "Base Python Image",
"default": os.getenv("SANDBOX_BASE_PYTHON_IMAGE", "infiniflow/sandbox-base-python:latest"),
"description": "Python runtime image used by executor-managed containers.",
"scope": "deployment",
"readonly": True,
},
"base_nodejs_image": {
"type": "string",
"required": False,
"label": "Base Node.js Image",
"default": os.getenv("SANDBOX_BASE_NODEJS_IMAGE", "infiniflow/sandbox-base-nodejs:latest"),
"description": "Node.js runtime image used by executor-managed containers.",
"scope": "deployment",
"readonly": True,
},
"executor_manager_port": {
"type": "integer",
"required": False,
"label": "Executor Manager Port",
"default": int(os.getenv("SANDBOX_EXECUTOR_MANAGER_PORT", "9385")),
"min": 1,
"max": 65535,
"description": "Host port exposed by sandbox-executor-manager.",
"scope": "deployment",
"readonly": True,
},
"enable_seccomp": {
"type": "boolean",
"required": False,
"label": "Enable Seccomp",
"default": os.getenv("SANDBOX_ENABLE_SECCOMP", "false").lower() == "true",
"description": "Whether sandbox-executor-manager starts containers with seccomp enabled.",
"scope": "deployment",
"readonly": True,
},
"max_memory": {
"type": "string",
"required": False,
"label": "Max Memory",
"default": os.getenv("SANDBOX_MAX_MEMORY", "256m"),
"description": "Memory limit applied to each sandbox container. Common format: 256m or 1g.",
"scope": "deployment",
"readonly": True,
},
"sandbox_timeout": {
"type": "string",
"required": False,
"label": "Sandbox Timeout",
"default": os.getenv("SANDBOX_TIMEOUT", "10s"),
"description": "Executor-manager container timeout for each sandbox run. Common format: 10s or 1m.",
"scope": "deployment",
"readonly": True,
},
}
def _normalize_language(self, language: str) -> str:
@@ -347,7 +393,7 @@ class SelfManagedProvider(SandboxProvider):
return False, f"Invalid endpoint format: {endpoint}. Must start with http:// or https://"
# Validate pool_size is positive
pool_size = config.get("pool_size", 10)
pool_size = config.get("executor_manager_pool_size", config.get("pool_size", 3))
if isinstance(pool_size, int) and pool_size <= 0:
return False, "Pool size must be greater than 0"

664
agent/sandbox/providers/ssh.py Normal file
View File

@@ -0,0 +1,664 @@
#
# Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from __future__ import annotations
import base64
import io
import json
import mimetypes
import os
import posixpath
import shlex
import stat
import time
import uuid
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from agent.sandbox.result_protocol import (
build_javascript_wrapper,
build_python_wrapper,
extract_structured_result,
)
from .base import (
ExecutionResult,
SandboxInstance,
SandboxProvider,
SandboxProviderConfigError,
)
if TYPE_CHECKING:
import paramiko
ALLOWED_ARTIFACT_EXTENSIONS = {
".csv",
".html",
".jpeg",
".jpg",
".json",
".pdf",
".png",
".svg",
}
class SSHProvider(SandboxProvider):
"""Execute code on a remote host through SSH."""
def __init__(self):
self.host = ""
self.port = 22
self.username = ""
self.password = ""
self.private_key = ""
self.passphrase = ""
self.python_bin = "python3"
self.node_bin = "node"
self.work_dir = "/tmp"
self.timeout = 30
self.max_output_bytes = 1024 * 1024
self.max_artifacts = 20
self.max_artifact_bytes = 10 * 1024 * 1024
self._initialized = False
self._instances: dict[str, dict[str, Any]] = {}
def initialize(self, config: Dict[str, Any]) -> bool:
self.host = str(config.get("host", "")).strip()
self.port = int(config.get("port", 22) or 22)
self.username = str(config.get("username", "")).strip()
self.password = str(config.get("password", "") or "")
self.private_key = str(config.get("private_key", "") or "")
self.passphrase = str(config.get("passphrase", "") or "")
self.python_bin = str(config.get("python_bin", "python3") or "python3").strip() or "python3"
self.node_bin = str(config.get("node_bin", "node") or "node").strip() or "node"
self.work_dir = str(config.get("work_dir", "/tmp") or "/tmp").strip() or "/tmp"
self.timeout = int(config.get("timeout", 30) or 30)
self.max_output_bytes = int(config.get("max_output_bytes", 1024 * 1024) or 1024 * 1024)
self.max_artifacts = int(config.get("max_artifacts", 20) or 20)
self.max_artifact_bytes = int(config.get("max_artifact_bytes", 10 * 1024 * 1024) or 10 * 1024 * 1024)
is_valid, error_message = self.validate_config(
{
"host": self.host,
"port": self.port,
"username": self.username,
"password": self.password,
"private_key": self.private_key,
"passphrase": self.passphrase,
"python_bin": self.python_bin,
"node_bin": self.node_bin,
"work_dir": self.work_dir,
"timeout": self.timeout,
"max_output_bytes": self.max_output_bytes,
"max_artifacts": self.max_artifacts,
"max_artifact_bytes": self.max_artifact_bytes,
}
)
if not is_valid:
raise SandboxProviderConfigError(error_message or "Invalid SSH provider configuration.")
self._assert_connectivity()
self._initialized = True
return True
def create_instance(self, template: str = "python") -> SandboxInstance:
if not self._initialized:
raise RuntimeError("Provider not initialized. Call initialize() first.")
language = self._normalize_language(template)
client = self._create_ssh_client()
sftp = client.open_sftp()
try:
remote_work_dir = self._create_remote_workspace(client)
stdout, stderr, exit_code = self._run_remote_command(
client,
f"mkdir -p {shlex.quote(posixpath.join(remote_work_dir, 'artifacts'))}",
timeout=min(self.timeout, 10),
)
if exit_code != 0:
raise RuntimeError(
f"Failed to create remote artifacts directory: {stderr or stdout or 'unknown error'}"
)
except Exception:
sftp.close()
client.close()
raise
instance_id = str(uuid.uuid4())
self._instances[instance_id] = {
"client": client,
"sftp": sftp,
"remote_work_dir": remote_work_dir,
"language": language,
}
return SandboxInstance(
instance_id=instance_id,
provider="ssh",
status="running",
metadata={"language": language, "remote_work_dir": remote_work_dir},
)
def execute_code(
self,
instance_id: str,
code: str,
language: str,
timeout: int = 10,
arguments: Optional[Dict[str, Any]] = None,
) -> ExecutionResult:
if not self._initialized:
raise RuntimeError("Provider not initialized. Call initialize() first.")
if instance_id not in self._instances:
raise RuntimeError(f"Unknown SSH sandbox instance: {instance_id}")
normalized_lang = self._normalize_language(language)
instance = self._instances[instance_id]
client: paramiko.SSHClient = instance["client"]
sftp: paramiko.SFTPClient = instance["sftp"]
remote_work_dir: str = instance["remote_work_dir"]
args_json = json.dumps(arguments or {}, ensure_ascii=False)
remote_script_path, command = self._upload_script(
sftp=sftp,
remote_work_dir=remote_work_dir,
language=normalized_lang,
code=code,
args_json=args_json,
)
requested_timeout = self.timeout if timeout is None else int(timeout)
if requested_timeout <= 0:
raise RuntimeError(f"Execution timeout must be greater than 0 seconds, got {requested_timeout}.")
exec_timeout = min(requested_timeout, self.timeout)
start_time = time.time()
stdout, stderr, exit_code = self._run_remote_command(client, command, timeout=exec_timeout)
execution_time = time.time() - start_time
self._validate_output_size(stdout, stderr)
stdout, structured_result = extract_structured_result(stdout)
return ExecutionResult(
stdout=stdout,
stderr=stderr,
exit_code=exit_code,
execution_time=execution_time,
metadata={
"instance_id": instance_id,
"language": normalized_lang,
"script_path": remote_script_path,
"remote_work_dir": remote_work_dir,
"status": "ok" if exit_code == 0 else "error",
"timeout": exec_timeout,
"command": command,
"artifacts": self._collect_artifacts(
sftp, posixpath.join(remote_work_dir, "artifacts")
),
"result_present": structured_result.get("present", False),
"result_value": structured_result.get("value"),
"result_type": structured_result.get("type"),
},
)
def destroy_instance(self, instance_id: str) -> bool:
if not self._initialized:
raise RuntimeError("Provider not initialized. Call initialize() first.")
if instance_id not in self._instances:
return True
instance = self._instances.pop(instance_id)
client: paramiko.SSHClient = instance["client"]
sftp: paramiko.SFTPClient = instance["sftp"]
remote_work_dir: str = instance["remote_work_dir"]
cleanup_error: Optional[Exception] = None
try:
stdout, stderr, exit_code = self._run_remote_command(
client,
f"rm -rf {shlex.quote(remote_work_dir)}",
timeout=min(self.timeout, 10),
)
if exit_code != 0:
raise RuntimeError(stderr or stdout or "unknown error")
except Exception as exc:
cleanup_error = exc
finally:
try:
sftp.close()
finally:
client.close()
if cleanup_error is not None:
raise RuntimeError(f"Failed to clean remote workspace {remote_work_dir}: {cleanup_error}")
return True
def health_check(self) -> bool:
try:
self._assert_connectivity()
return True
except Exception:
return False
def _assert_connectivity(self) -> None:
try:
client = self._create_ssh_client()
try:
_, stderr, exit_code = self._run_remote_command(
client,
"true",
timeout=min(self.timeout, 10),
)
if exit_code != 0:
raise SandboxProviderConfigError(
f"SSH connectivity check failed on {self.username}@{self.host}:{self.port}: "
f"{stderr or 'remote command returned non-zero exit status'}"
)
finally:
client.close()
except SandboxProviderConfigError:
raise
except Exception as exc:
raise SandboxProviderConfigError(
f"Failed to connect to SSH host {self.username}@{self.host}:{self.port}: {exc}"
) from exc
def get_supported_languages(self) -> List[str]:
return ["python", "javascript", "nodejs"]
@staticmethod
def get_config_schema() -> Dict[str, Dict]:
return {
"host": {
"type": "string",
"required": True,
"label": "SSH Host",
"placeholder": "192.168.1.10",
"description": "Remote host that will execute generated code.",
},
"port": {
"type": "integer",
"required": True,
"label": "SSH Port",
"default": 22,
"min": 1,
"max": 65535,
"description": "SSH port on the remote host.",
},
"username": {
"type": "string",
"required": True,
"label": "SSH Username",
"placeholder": "ragflow",
"description": "Username used to connect to the remote host.",
},
"password": {
"type": "string",
"required": False,
"label": "SSH Password",
"secret": True,
"placeholder": "Optional when using a private key",
"description": "Password-based SSH authentication.",
},
"private_key": {
"type": "string",
"required": False,
"label": "SSH Private Key",
"secret": True,
"multiline": True,
"placeholder": "Paste PEM content or enter a local file path",
"description": "Private key PEM content or a readable private key path on the RAGFlow host.",
},
"passphrase": {
"type": "string",
"required": False,
"label": "Private Key Passphrase",
"secret": True,
"placeholder": "Optional",
"description": "Passphrase for the private key if it is encrypted.",
},
"python_bin": {
"type": "string",
"required": False,
"default": "python3",
"label": "Python Binary",
"description": "Python executable used for remote code execution.",
},
"node_bin": {
"type": "string",
"required": False,
"default": "node",
"label": "Node.js Binary",
"description": "Node.js executable used for remote JavaScript execution.",
},
"work_dir": {
"type": "string",
"required": False,
"label": "Remote Workspace Root",
"default": "/tmp",
"placeholder": "/tmp",
"description": "Writable remote directory used to create a temporary workspace.",
},
"timeout": {
"type": "integer",
"required": False,
"label": "Timeout (seconds)",
"default": 30,
"min": 1,
"max": 600,
"description": "Maximum SSH execution time for a single run.",
},
"max_output_bytes": {
"type": "integer",
"required": False,
"label": "Max Output Bytes",
"default": 1048576,
"min": 1024,
"max": 10485760,
"description": "Maximum combined stdout and stderr size.",
},
"max_artifacts": {
"type": "integer",
"required": False,
"label": "Max Artifacts",
"default": 20,
"min": 0,
"max": 100,
"description": "Maximum number of files collected from the remote artifacts directory.",
},
"max_artifact_bytes": {
"type": "integer",
"required": False,
"label": "Max Artifact Bytes",
"default": 10485760,
"min": 1024,
"max": 104857600,
"description": "Maximum size of a single artifact file in bytes.",
},
}
def validate_config(self, config: Dict[str, Any]) -> tuple[bool, Optional[str]]:
host = str(config.get("host", "") or "").strip()
username = str(config.get("username", "") or "").strip()
password = str(config.get("password", "") or "")
private_key = str(config.get("private_key", "") or "")
python_bin = str(config.get("python_bin", "python3") or "python3").strip()
node_bin = str(config.get("node_bin", "node") or "node").strip()
if not host:
return False, "SSH host is required"
if not username:
return False, "SSH username is required"
if not password and not private_key:
return False, "Either password or private_key must be provided"
if not python_bin:
return False, "Python binary is required"
if not node_bin:
return False, "Node.js binary is required"
try:
port = int(config.get("port", 22) or 22)
except (TypeError, ValueError):
return False, "SSH port must be an integer"
if port <= 0 or port > 65535:
return False, "SSH port must be between 1 and 65535"
for key in ("timeout", "max_output_bytes", "max_artifacts", "max_artifact_bytes"):
try:
value = int(config.get(key, 0) or 0)
except (TypeError, ValueError):
return False, f"{key} must be an integer"
if key == "max_artifacts":
if value < 0:
return False, "max_artifacts must be greater than or equal to 0"
elif value <= 0:
return False, f"{key} must be greater than 0"
return True, None
def _create_ssh_client(self) -> paramiko.SSHClient:
paramiko = _get_paramiko_module()
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
connect_kwargs: dict[str, Any] = {
"hostname": self.host,
"port": self.port,
"username": self.username,
"timeout": self.timeout,
"banner_timeout": self.timeout,
"auth_timeout": self.timeout,
"look_for_keys": False,
"allow_agent": False,
}
if self.private_key:
connect_kwargs["pkey"] = self._load_private_key()
if self.password:
connect_kwargs["password"] = self.password
client.connect(**connect_kwargs)
return client
def _load_private_key(self) -> paramiko.PKey:
paramiko = _get_paramiko_module()
loaders = (
paramiko.RSAKey,
paramiko.Ed25519Key,
paramiko.ECDSAKey,
paramiko.DSSKey,
)
errors: list[str] = []
private_key_value = self.private_key.strip()
passphrase = self.passphrase or None
if os.path.exists(private_key_value):
for key_cls in loaders:
try:
return key_cls.from_private_key_file(private_key_value, password=passphrase)
except Exception as exc:
errors.append(str(exc))
else:
for key_cls in loaders:
try:
return key_cls.from_private_key(io.StringIO(private_key_value), password=passphrase)
except Exception as exc:
errors.append(str(exc))
raise SandboxProviderConfigError(
"Failed to load SSH private key. " + "; ".join(error for error in errors if error)
)
def _create_remote_workspace(self, client: paramiko.SSHClient) -> str:
base_dir = self.work_dir.rstrip("/") or "/tmp"
template = posixpath.join(base_dir, "ragflow-codeexec.XXXXXX")
stdout, stderr, exit_code = self._run_remote_command(
client,
f"mkdir -p {shlex.quote(base_dir)} && mktemp -d {shlex.quote(template)}",
timeout=min(self.timeout, 10),
)
if exit_code != 0:
raise RuntimeError(
f"Failed to create remote workspace on {self.host}: {stderr or stdout or 'unknown error'}"
)
remote_work_dir = stdout.strip().splitlines()[-1] if stdout.strip() else ""
if not remote_work_dir:
raise RuntimeError("Remote workspace creation did not return a path.")
return remote_work_dir
def _upload_script(
self,
sftp: paramiko.SFTPClient,
remote_work_dir: str,
language: str,
code: str,
args_json: str,
) -> tuple[str, str]:
if language == "python":
script_name = "main.py"
script_content = build_python_wrapper(code, args_json)
elif language in {"javascript", "nodejs"}:
script_name = "main.js"
script_content = build_javascript_wrapper(code, args_json)
else:
raise RuntimeError(f"Unsupported language for SSH provider: {language}")
remote_script_path = posixpath.join(remote_work_dir, script_name)
with sftp.file(remote_script_path, "w") as remote_file:
remote_file.write(script_content)
command = self._build_execution_command(remote_work_dir, remote_script_path, language)
return remote_script_path, command
def _build_execution_command(self, remote_work_dir: str, remote_script_path: str, language: str) -> str:
normalized_lang = self._normalize_language(language)
if normalized_lang == "python":
executable = self.python_bin
elif normalized_lang == "nodejs":
executable = self.node_bin
else:
raise RuntimeError(f"Unsupported language for SSH provider: {language}")
return (
f"cd {shlex.quote(remote_work_dir)} && "
f"{shlex.quote(executable)} {shlex.quote(remote_script_path)}"
)
def _run_remote_command(
self,
client: paramiko.SSHClient,
command: str,
timeout: int,
) -> tuple[str, str, int]:
stdin, stdout_stream, stderr_stream = client.exec_command(command, timeout=timeout)
stdin.close()
channel = stdout_stream.channel
stdout_chunks: list[bytes] = []
stderr_chunks: list[bytes] = []
deadline = time.time() + timeout
while True:
while channel.recv_ready():
stdout_chunks.append(channel.recv(65536))
while channel.recv_stderr_ready():
stderr_chunks.append(channel.recv_stderr(65536))
if channel.exit_status_ready():
break
if time.time() > deadline:
channel.close()
raise TimeoutError(f"Execution timed out after {timeout} seconds")
time.sleep(0.1)
while channel.recv_ready():
stdout_chunks.append(channel.recv(65536))
while channel.recv_stderr_ready():
stderr_chunks.append(channel.recv_stderr(65536))
exit_code = channel.recv_exit_status()
stdout = b"".join(stdout_chunks).decode("utf-8", errors="replace")
stderr = b"".join(stderr_chunks).decode("utf-8", errors="replace")
return stdout, stderr, exit_code
def _validate_output_size(self, stdout: str, stderr: str) -> None:
output_size = len((stdout or "").encode("utf-8")) + len((stderr or "").encode("utf-8"))
if output_size > self.max_output_bytes:
raise RuntimeError(f"SSH execution output exceeded {self.max_output_bytes} bytes.")
def _collect_artifacts(
self,
sftp: paramiko.SFTPClient,
artifacts_dir: str,
) -> list[dict[str, Any]]:
artifacts: list[dict[str, Any]] = []
self._collect_artifacts_recursive(sftp, artifacts_dir, "", artifacts)
return artifacts
def _collect_artifacts_recursive(
self,
sftp: paramiko.SFTPClient,
current_dir: str,
relative_dir: str,
artifacts: list[dict[str, Any]],
) -> None:
try:
entries = sftp.listdir_attr(current_dir)
except FileNotFoundError:
return
for entry in sorted(entries, key=lambda item: item.filename):
name = entry.filename
remote_path = posixpath.join(current_dir, name)
relative_path = posixpath.join(relative_dir, name) if relative_dir else name
mode = entry.st_mode
if mode is None:
mode = sftp.lstat(remote_path).st_mode
if mode is None:
raise RuntimeError(f"Unable to determine artifact entry type: {relative_path}")
if stat.S_ISLNK(mode):
raise RuntimeError(f"Artifact symlinks are not allowed: {relative_path}")
if stat.S_ISDIR(mode):
self._collect_artifacts_recursive(sftp, remote_path, relative_path, artifacts)
continue
if not stat.S_ISREG(mode):
raise RuntimeError(f"Unsupported artifact entry: {relative_path}")
if len(artifacts) >= self.max_artifacts:
raise RuntimeError(f"SSH execution produced more than {self.max_artifacts} artifacts.")
size = int(entry.st_size or 0)
if size > self.max_artifact_bytes:
raise RuntimeError(f"Artifact exceeds {self.max_artifact_bytes} bytes: {relative_path}")
ext = os.path.splitext(name)[1].lower()
if ext not in ALLOWED_ARTIFACT_EXTENSIONS:
raise RuntimeError(f"Unsupported artifact type: {relative_path}")
with sftp.file(remote_path, "rb") as artifact_file:
content = artifact_file.read()
artifacts.append(
{
"name": relative_path,
"content_b64": base64.b64encode(content).decode("ascii"),
"mime_type": mimetypes.guess_type(name)[0] or "application/octet-stream",
"size": size,
}
)
@staticmethod
def _normalize_language(language: str) -> str:
lang_lower = (language or "python").lower()
if lang_lower in {"python", "python3"}:
return "python"
if lang_lower in {"javascript", "nodejs"}:
return "nodejs"
return lang_lower
def _get_paramiko_module():
try:
import paramiko
except ImportError as exc:
raise SandboxProviderConfigError(
"paramiko is required for the SSH sandbox provider. Install the project dependencies to enable it."
) from exc
return paramiko

3
agent/tools/base.py
View File

@@ -19,6 +19,7 @@ import time
from copy import deepcopy
import asyncio
from functools import partial
from collections.abc import Mapping
from typing import TypedDict, List, Any
from agent.component.base import ComponentParamBase, ComponentBase
from common.misc_utils import hash_str2int
@@ -58,6 +59,8 @@ class LLMToolPluginCallSession(ToolCallSession):
async def tool_call_async(self, name: str, arguments: dict[str, Any], request_timeout: float | int = 10) -> Any:
assert name in self.tools_map, f"LLM tool {name} does not exist"
logging.info(f"[ToolCall] invoke name={name} arguments={str(arguments)[:200]}")
if not isinstance(arguments, Mapping):
raise TypeError(f"Tool arguments for {name} must be an object, got {type(arguments).__name__}")
st = timer()
tool_obj = self.tools_map[name]
if isinstance(tool_obj, MCPToolBinding):

27
agent/tools/code_exec.py
View File

@@ -361,11 +361,21 @@ class CodeExec(ToolBase, ABC):
# Try using the new sandbox provider system first
try:
from agent.sandbox.client import execute_code as sandbox_execute_code
from agent.sandbox.client import get_provider_info
from agent.sandbox.client import reload_provider
from agent.sandbox.providers.base import SandboxProviderConfigError
if self.check_if_canceled("CodeExec execution"):
return
reload_provider()
provider_info = get_provider_info()
provider_type = provider_info.get("provider_type") or "unknown"
logging.info(
f"[CodeExec]: dispatching execution to sandbox provider '{provider_type}' "
f"(language={language}, timeout={timeout_seconds}s)"
)
# Execute code using the provider system
result = sandbox_execute_code(code=code, language=language, timeout=timeout_seconds, arguments=arguments)
@@ -376,7 +386,7 @@ class CodeExec(ToolBase, ABC):
return self._process_execution_result(
result.stdout,
result.stderr,
"Provider system",
f"Provider system ({provider_type})",
artifacts,
execution_metadata=result.metadata,
)
@@ -388,10 +398,8 @@ class CodeExec(ToolBase, ABC):
# Provider modules are unavailable, fall back to legacy HTTP sandbox.
logging.info(f"[CodeExec]: Provider system not available, using HTTP fallback: {provider_error}")
except RuntimeError as provider_error:
if not self._should_fallback_to_http(provider_error):
self.set_output("_ERROR", f"Provider system execution failed: {provider_error}")
return self.output()
logging.info(f"[CodeExec]: Provider system not available, using HTTP fallback: {provider_error}")
self.set_output("_ERROR", f"Provider system execution failed: {provider_error}")
return self.output()
# Fallback to direct HTTP request
code_b64 = self._encode_code(code)
@@ -502,15 +510,6 @@ class CodeExec(ToolBase, ABC):
return metadata.get("result_value"), False
return self._deserialize_stdout(stdout), True
@staticmethod
def _should_fallback_to_http(provider_error: RuntimeError) -> bool:
message = str(provider_error).lower()
fallback_markers = (
"no sandbox provider configured",
"sandbox provider type not configured",
)
return any(marker in message for marker in fallback_markers)
@classmethod
def _ensure_bucket_lifecycle(cls):
if cls._lifecycle_configured: