kodeagent.code_runner#


Run Python code generated by an LLM in a given environment. Currently, supports local host and E2B sandbox execution. Also, defines custom exceptions for security and environment errors.

User Code

[1] Syntax Check (AST parse)

[2] Import Whitelist

[3] Pattern Detection - Catches obfuscation

[4] LLM Review

[5] Execute Code (locally or in Sandbox)

CodeRunResult

Named tuple for code run results.

CodeRunner

Run Python code generated by an LLM in a given environment.

CodeRunnerEnv

Abstract base class for code execution environments.

E2BCodeRunnerEnv

Execution environment for the E2B sandbox.

HostCodeRunnerEnv

Execution environment for the local host.

Run Python code generated by an LLM in a given environment. Currently, supports local host and E2B sandbox execution. Also, defines custom exceptions for security and environment errors.

User Code

[1] Syntax Check (AST parse)

[2] Import Whitelist

[3] Pattern Detection - Catches obfuscation

[4] LLM Review

[5] Execute Code (locally or in Sandbox)

kodeagent.code_runner.CODE_ENV_NAMES#

Allowed code execution environment names.

alias of Literal[‘host’, ‘docker’, ‘e2b’]

class kodeagent.code_runner.CodeRunResult(stdout: str, stderr: str, return_code: int, generated_files: list[str])[source]#

Bases: NamedTuple

Named tuple for code run results.

Create new instance of CodeRunResult(stdout, stderr, return_code, generated_files)

generated_files: list[str]#

Alias for field number 3

return_code: int#

Alias for field number 2

stderr: str#

Alias for field number 1

stdout: str#

Alias for field number 0

class kodeagent.code_runner.CodeRunner(env: Literal['host', 'docker', 'e2b'], allowed_imports: list[str], model_name: str, pip_packages: str | None = None, timeout: int = 30, env_vars_to_set: dict[str, str] | None = None, litellm_params: dict | None = None, work_dir: str | None = None, usage_tracker: UsageTracker | None = None, tool_names: set[str] | None = None)[source]#

Bases: object

Run Python code generated by an LLM in a given environment.

Create an environment to run Python code.

Parameters:
  • env – The code execution environment. Must be a string from CODE_ENV_NAMES.

  • allowed_imports – A list of Python modules that are allowed to be imported.

  • model_name – The LLM model name to use for security review.

  • pip_packages – Optional Python libs to be installed by pip [E2B].

  • timeout – Code execution timeout (default 30s).

  • env_vars_to_set – Optional environment variables to set in the code execution environment (E2B only).

  • litellm_params – Optional parameters for LiteLLM.

  • work_dir – Optional local workspace directory.

  • usage_tracker – Optional UsageTracker instance.

  • tool_names – Optional set of whitelisted tool names provided by the user.

check_imports(code: str) set[str][source]#

Check for disallowed imports in code, allowing submodules.

Parameters:

code – The Python source code to check.

Returns:

A set of disallowed import module names found in the code.

Raises:

CodeSecurityError – If dangerous builtins are used.

cleanup()[source]#

Clean up resources in the environment.

async download_files_from_remote(remote_paths: list[str]) list[str][source]#

Download files from the remote environment to the local workspace.

Parameters:

remote_paths – List of absolute paths in the remote environment.

Returns:

List of local absolute paths for the downloaded files.

property local_modules_to_copy: list[str]#

Get the list of local modules to copy to the execution environment.

Returns:

A list of local module filenames.

async run(tools_code: str, generated_code: str, task_id: str) CodeRunResult[source]#

Run Python code in pre-specified environment after security review.

Parameters:
  • tools_code – The Python source code for agent tools.

  • generated_code – The Python source code generated to solve a task.

  • task_id – Unique task identifier for tracking.

Returns:

A tuple of (stdout, stderr, return_code, generated_files).

Raises:
class kodeagent.code_runner.CodeRunnerEnv(work_dir: str | None = None)[source]#

Bases: ABC

Abstract base class for code execution environments.

Initialize the code runner environment.

Parameters:

work_dir – Optional local workspace directory. Output files from code execution will be stored here. It will be used only if the path already exists. Otherwise, a temporary directory will be created.

cleanup()[source]#

Clean up environment resources. Implementation is left to the subclass if required. Cleaning up resources may have a side effect of removing the files produced by the code.

abstract async download_files_from_remote(remote_paths: list[str]) list[str][source]#

Download files from the environment to the local work_dir.

property effective_work_dir: str#

Return the effective working directory specified or creating a temporary one if needed.

Returns:

The effective working directory path.

abstract async run(source_code: str, task_id: str, timeout: int) CodeRunResult[source]#

Execute Python code in the environment.

Parameters:
  • source_code – The Python source code to execute.

  • task_id – Unique task identifier.

  • timeout – Timeout for code execution.

Returns:

A tuple of (stdout, stderr, return_code, generated_files).

exception kodeagent.code_runner.CodeSecurityError[source]#

Bases: Exception

Exception raised for security violations in code execution.

kodeagent.code_runner.DANGEROUS_BUILTINS = {'__import__', 'compile', 'eval', 'exec'}#

Dangerous built-in functions that are not allowed in code execution.

kodeagent.code_runner.DEFAULT_ALLOWED_IMPORTS = {'ast', 'operator', 'random', 're', 'time'}#

Default allowed imports for code execution.

class kodeagent.code_runner.E2BCodeRunnerEnv(work_dir: str | None = None, env_vars: dict[str, str] | None = None, pip_packages_str: str | None = None)[source]#

Bases: CodeRunnerEnv

Execution environment for the E2B sandbox.

Initialize the E2B code runner environment.

Parameters:
  • work_dir – Optional local workspace directory. Output files from code execution will be stored here. It will be used only if the path already exists. Otherwise, a temporary directory will be created.

  • env_vars – Optional environment variables to set in the E2B sandbox.

  • pip_packages_str – Optional string of pip packages to install in the E2B sandbox.

cleanup()[source]#

Close the sandbox and clean up.

async download_files_from_remote(remote_paths: list[str]) list[str][source]#

Download files from the E2B sandbox to the local work_dir.

Parameters:

remote_paths – List of absolute paths in the E2B sandbox.

Returns:

List of local absolute paths for the downloaded files.

async run(source_code: str, task_id: str, timeout: int) CodeRunResult[source]#

Execute Python code in the E2B sandbox.

Parameters:
  • source_code – The Python source code to execute.

  • task_id – Unique task identifier.

  • timeout – Timeout for code execution.

Returns:

A tuple of (stdout, stderr, return_code, generated_files).

class kodeagent.code_runner.HostCodeRunnerEnv(work_dir: str | None = None)[source]#

Bases: CodeRunnerEnv

Execution environment for the local host.

Initialize the code runner environment.

Parameters:

work_dir – Optional local workspace directory. Output files from code execution will be stored here. It will be used only if the path already exists. Otherwise, a temporary directory will be created.

async download_files_from_remote(remote_paths: list[str]) list[str][source]#

On host, files are already local.

Parameters:

remote_paths – List of absolute paths in the host environment.

Returns:

The same list of paths, as they are already local.

async run(source_code: str, task_id: str, timeout: int) CodeRunResult[source]#

Execute Python code on the local host.

Parameters:
  • source_code – The Python source code to execute.

  • task_id – Unique task identifier.

  • timeout – Timeout for code execution.

Returns:

A tuple of (stdout, stderr, return_code, generated_files).

exception kodeagent.code_runner.UnknownCodeEnvError[source]#

Bases: Exception

Exception raised for unknown code execution environments.