# Generated by SnakeBridge v0.13.0 - DO NOT EDIT MANUALLY # Regenerate with: mix compile # Library: dspy 3.1.2 # Python module: dspy # Python class: PythonInterpreter defmodule Dspy.PythonInterpreter do @moduledoc """ Local interpreter for secure Python execution using Deno and Pyodide. Implements the Interpreter protocol for secure code execution in a WASM-based sandbox. Code runs in an isolated Pyodide environment with no access to the host filesystem, network, or environment by default. Prerequisites: Deno must be installed: https://docs.deno.com/runtime/getting_started/installation/ ## Examples ```python # Basic execution with PythonInterpreter() as interp: result = interp("print(1 + 2)") # Returns "3" # With host-side tools def my_tool(question: str) -> str: return "answer" with PythonInterpreter(tools={"my_tool": my_tool}) as interp: result = interp("print(my_tool(question='test'))") ``` """ def __snakebridge_python_name__, do: "dspy" def __snakebridge_python_class__, do: "PythonInterpreter" def __snakebridge_library__, do: "dspy" @opaque t :: SnakeBridge.Ref.t() @doc """ Args: deno_command: command list to launch Deno. enable_read_paths: Files or directories to allow reading from in the sandbox. enable_write_paths: Files or directories to allow writing to in the sandbox. All write paths will also be able to be read from for mounting. enable_env_vars: Environment variable names to allow in the sandbox. enable_network_access: Domains or IPs to allow network access in the sandbox. sync_files: If set, syncs changes within the sandbox back to original files after execution. tools: Dictionary mapping tool names to callable functions. Each function should accept keyword arguments and return a string. Tools are callable directly from sandbox code by name. output_fields: List of output field definitions for typed SUBMIT signature. Each dict should have 'name' and optionally 'type' keys. ## Parameters - `deno_command` (term() default: None) - `enable_read_paths` (term() default: None) - `enable_write_paths` (term() default: None) - `enable_env_vars` (term() default: None) - `enable_network_access` (term() default: None) - `sync_files` (boolean() default: True) - `tools` (term() default: None) - `output_fields` (term() default: None) """ @spec new(list(term()), keyword()) :: {:ok, SnakeBridge.Ref.t()} | {:error, Snakepit.Error.t()} def new(args, opts \\ []) do {args, opts} = SnakeBridge.Runtime.normalize_args_opts(args, opts) SnakeBridge.Runtime.call_class(__MODULE__, :__init__, [] ++ List.wrap(args), opts) end @doc """ Python method `PythonInterpreter._ensure_deno_process`. ## Returns - `nil` """ @spec _ensure_deno_process(SnakeBridge.Ref.t(), keyword()) :: {:ok, nil} | {:error, Snakepit.Error.t()} def _ensure_deno_process(ref, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_ensure_deno_process, [], opts) end @doc """ Extract parameter info from a callable for sandbox registration. ## Parameters - `py_fn` (term()) ## Returns - `list(%{optional(term()) => term()})` """ @spec _extract_parameters(SnakeBridge.Ref.t(), term(), keyword()) :: {:ok, list(%{optional(term()) => term()})} | {:error, Snakepit.Error.t()} def _extract_parameters(ref, py_fn, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_extract_parameters, [py_fn], opts) end @doc """ Python method `PythonInterpreter._get_deno_dir`. ## Returns - `term()` """ @spec _get_deno_dir(SnakeBridge.Ref.t(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def _get_deno_dir(ref, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_get_deno_dir, [], opts) end @doc """ Python method `PythonInterpreter._get_runner_path`. ## Returns - `String.t()` """ @spec _get_runner_path(SnakeBridge.Ref.t(), keyword()) :: {:ok, String.t()} | {:error, Snakepit.Error.t()} def _get_runner_path(ref, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_get_runner_path, [], opts) end @doc """ Handle a tool call request from the sandbox. ## Parameters - `request` (%{optional(term()) => term()}) ## Returns - `nil` """ @spec _handle_tool_call(SnakeBridge.Ref.t(), %{optional(term()) => term()}, keyword()) :: {:ok, nil} | {:error, Snakepit.Error.t()} def _handle_tool_call(ref, request, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_handle_tool_call, [request], opts) end @doc """ Insert Python assignments for each variable at the top of the code. ## Parameters - `code` (String.t()) - `variables` (%{optional(String.t()) => term()}) ## Returns - `String.t()` """ @spec _inject_variables( SnakeBridge.Ref.t(), String.t(), %{optional(String.t()) => term()}, keyword() ) :: {:ok, String.t()} | {:error, Snakepit.Error.t()} def _inject_variables(ref, code, variables, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_inject_variables, [code, variables], opts) end @doc """ Python method `PythonInterpreter._mount_files`. ## Returns - `term()` """ @spec _mount_files(SnakeBridge.Ref.t(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def _mount_files(ref, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_mount_files, [], opts) end @doc """ Register tools and output fields with the sandbox. ## Returns - `nil` """ @spec _register_tools(SnakeBridge.Ref.t(), keyword()) :: {:ok, nil} | {:error, Snakepit.Error.t()} def _register_tools(ref, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_register_tools, [], opts) end @doc """ Serialize a Python value to a string representation for injection. ## Parameters - `value` (term()) ## Returns - `String.t()` """ @spec _serialize_value(SnakeBridge.Ref.t(), term(), keyword()) :: {:ok, String.t()} | {:error, Snakepit.Error.t()} def _serialize_value(ref, value, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_serialize_value, [value], opts) end @doc """ Python method `PythonInterpreter._sync_files`. ## Returns - `term()` """ @spec _sync_files(SnakeBridge.Ref.t(), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def _sync_files(ref, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :_sync_files, [], opts) end @doc """ Python method `PythonInterpreter.execute`. ## Parameters - `code` (String.t()) - `variables` (term() default: None) ## Returns - `term()` """ @spec execute(SnakeBridge.Ref.t(), String.t(), list(term()), keyword()) :: {:ok, term()} | {:error, Snakepit.Error.t()} def execute(ref, code, args, opts \\ []) do {args, opts} = SnakeBridge.Runtime.normalize_args_opts(args, opts) SnakeBridge.Runtime.call_method(ref, :execute, [code] ++ List.wrap(args), opts) end @doc """ Python method `PythonInterpreter.shutdown`. ## Returns - `nil` """ @spec shutdown(SnakeBridge.Ref.t(), keyword()) :: {:ok, nil} | {:error, Snakepit.Error.t()} def shutdown(ref, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :shutdown, [], opts) end @doc """ Initialize the Deno/Pyodide sandbox. This pre-warms the sandbox by starting the Deno subprocess. Can be called explicitly for pooling, or will be called lazily on first execute(). Idempotent: safe to call multiple times. ## Returns - `nil` """ @spec start(SnakeBridge.Ref.t(), keyword()) :: {:ok, nil} | {:error, Snakepit.Error.t()} def start(ref, opts \\ []) do SnakeBridge.Runtime.call_method(ref, :start, [], opts) end end