code_exec.py

Overview

The code_exec.py file implements a sandboxed code execution tool component for the InfiniFlow system. It enables safe execution of user-supplied code snippets written in Python or JavaScript, returning structured results in JSON format. This functionality is encapsulated in a reusable component that validates, encodes, sends, and processes code execution requests through a remote sandbox service.

Key features include:

• Base64 encoding and validation of code to ensure safe transmission.

• Support for two programming languages: Python and Node.js (JavaScript).

• Structured input validation using Pydantic models.

• Integration with a remote sandbox service via HTTP POST requests.

• Parsing and mapping of execution results to defined outputs.

• Timeout control on code execution to prevent indefinite runs.

This file primarily contains data models to represent execution requests, parameter definitions for the tool, and the main tool class implementing the execution logic.

Classes and Functions

Enum: Language

class Language(StrEnum):
    PYTHON = "python"
    NODEJS = "nodejs"

• Purpose: Defines supported programming languages for code execution as string enumerations.

• Values:

  • "python": Python language.

  • "nodejs": Node.js (JavaScript) language.

Class: CodeExecutionRequest

class CodeExecutionRequest(BaseModel):
    code_b64: str
    language: str = Language.PYTHON.value
    arguments: Optional[dict] = {}

• Base class: pydantic.BaseModel

• Description: Represents a validated request to execute a piece of code in the sandbox.

• Fields:

  • code_b64 (str): Base64-encoded source code string. Must be valid base64.

  • language (str): Programming language of the code. Defaults to "python".

  • arguments (Optional[dict]): Optional dictionary of input arguments for the code.

Validators:

• validate_base64(cls, v: str) -> str

  • Validates that code_b64 is a properly base64-encoded string.

  • Raises ValueError if validation fails.

• normalize_language(cls, v) -> str

  • Normalizes input language strings to "python" or "nodejs".

  • Accepts variations like "python3" or "javascript".

  • Raises ValueError if an unsupported language is provided.

Usage Example

req = CodeExecutionRequest(
    code_b64=base64.b64encode(b"def main(): return {'result': 42}").decode(),
    language="python",
    arguments={"arg1": "value1"}
)

Class: CodeExecParam

class CodeExecParam(ToolParamBase):

• Base class: ToolParamBase

• Description: Defines the parameters and metadata for the code execution tool component.

• Properties:

  • meta (ToolMeta dict): Metadata including name, description, and parameters schema.

  • lang (str): Programming language. Default is "python".

  • script (str): The script code to execute; must include a main function.

  • arguments (dict): Input arguments for the script.

  • outputs (dict): Expected outputs from the script execution.

Important Details

• The meta contains detailed descriptions and example code snippets for Python and JavaScript usage.

• The check() method validates the parameters:

  • Ensures the language is one of the supported values.

  • Ensures the script is not empty.

• The get_input_form() method returns a dictionary describing the input fields required, derived from the arguments.

Usage Example

param = CodeExecParam()
param.lang = "python"
param.script = """
def main() -> dict:
    return {"result": "Hello, World!"}
"""
param.arguments = {}
param.outputs = {"result": {"value": "", "type": "string"}}
param.check()

Class: CodeExec

class CodeExec(ToolBase, ABC):

• Base classes: ToolBase, ABC

• Description: The main tool component that handles code execution requests by interacting with the sandbox service.

Properties

• component_name (class attribute): "CodeExec"

Methods

• _invoke(self, **kwargs)

  • Decorated with a timeout (default 10 minutes or environment variable COMPONENT_EXEC_TIMEOUT).

  • Retrieves language, script, and arguments from parameters or provided kwargs.

  • Resolves argument values from the canvas variables if not explicitly provided.

  • Calls _execute_code() to perform the actual code execution.

• _execute_code(self, language: str, code: str, arguments: dict)

  • Encodes the code with base64.

  • Constructs a CodeExecutionRequest.

  • Sends a POST request to the sandbox service at http://{SANDBOX_HOST}:9385/run.

  • Handles HTTP errors and response parsing.

  • Extracts and processes standard output (stdout) and standard error (stderr).

  • Maps the returned results to the output fields defined in the parameters.

  • Sets an _ERROR output key on errors.

  • Returns the outputs after execution.

• _encode_code(self, code: str) -> str

  • Helper method to encode a UTF-8 code string to base64.

• thoughts(self) -> str

  • Returns a short descriptive string about the operation being performed.

Usage Example

code_exec = CodeExec()
result = code_exec._execute_code(
    language="python",
    code="def main(): return {'result': 123}",
    arguments={}
)
print(result)

Important Implementation Details

• Sandbox Communication: The tool communicates with a remote sandbox service via HTTP POST, sending a JSON payload containing the base64-encoded code, language, and arguments.

• Timeout Control: Execution requests are wrapped with a timeout decorator to prevent long-running or stuck executions.

• Output Handling: The tool handles multiple possible output formats:

  • Tuple: maps tuple elements to outputs in order.

  • Dict: maps dictionary keys to output names.

  • Single value: sets the value to all output fields except those starting with an underscore.

• Error Management: Any errors in encoding, network communication, or code execution errors from the sandbox are captured and set as an _ERROR output.

• Input Validation: Uses Pydantic models and validators to ensure request integrity before sending.

Interaction with Other System Components

• Sandbox Service: Relies on an external sandbox service running at http://{SANDBOX_HOST}:9385/run to execute the code securely.

• Canvas Environment: Retrieves variable values from the _canvas object for arguments if not provided explicitly.

• Agent Tools Base: Inherits from base classes ToolBase and ToolParamBase, integrating into a larger agent/tool framework.

• Settings and Environment: Reads environment variables and settings (e.g., COMPONENT_EXEC_TIMEOUT, SANDBOX_HOST) for configuration.

• Logging: Uses Python’s logging module to record requests and responses for monitoring.

Visual Diagram

classDiagram
    class Language {
        <<enum>>
        +PYTHON = "python"
        +NODEJS = "nodejs"
    }

    class CodeExecutionRequest {
        +code_b64: str
        +language: str
        +arguments: Optional[dict]
        +validate_base64(v: str) str
        +normalize_language(v) str
    }

    class CodeExecParam {
        +meta: ToolMeta
        +lang: str
        +script: str
        +arguments: dict
        +outputs: dict
        +check()
        +get_input_form() dict
    }

    class CodeExec {
        +component_name: str = "CodeExec"
        +_invoke(**kwargs)
        +_execute_code(language: str, code: str, arguments: dict)
        +_encode_code(code: str) str
        +thoughts() str
    }

    CodeExecParam <|-- CodeExec
    CodeExecutionRequest ..> CodeExec : uses for request payload
    Language <.. CodeExecutionRequest : language field

Summary

code_exec.py is a critical utility within the InfiniFlow system that allows for dynamic, sandboxed execution of Python and JavaScript code snippets. It ensures safety, validation, and structured integration with the rest of the agent/tool framework by:

• Defining strict input data models.

• Encoding and securely transmitting code.

• Handling execution results and errors gracefully.

• Facilitating extensibility with clear parameter metadata.

This design enables users and automated agents to run arbitrary code fragments safely in a controlled environment, supporting complex data processing workflows within the InfiniFlow platform.