dspy.PythonInterpreter¶

def __call__(
    self,
    code: str,
    variables: dict[str, Any] | None = None,
) -> Any:
    return self.execute(code, variables)

@with_callbacks
def execute(
    self,
    code: str,
    variables: dict[str, Any] | None = None,
) -> Any:
    if self._handling_tool_call:
        raise CodeInterpreterError("PythonInterpreter cannot execute recursively from one of its tools.")
    self._check_session_active()
    self._check_thread_ownership()
    variables = variables or {}
    code = self._inject_variables(code, variables)
    self._ensure_deno_process()
    self._mount_files()
    self._register_tools()

    for name, value in self._pending_large_vars.items():
        self._inject_large_var(name, value)

    # Send the code as JSON-RPC request
    execute_request_id = self._next_request_id()
    input_data = _jsonrpc_request("execute", {"code": code}, execute_request_id)
    self._write_message(input_data, "during execution")

    # Read and handle messages until we get the final output.
    # Loop is needed because tool calls require back-and-forth communication.
    skipped = 0
    while skipped <= self._MAX_SKIP_LINES:
        output_line = self._read_response_line("during execution")
        msg = self._parse_response_line(output_line, "during execution")
        if msg is None:
            skipped += 1
            continue

        # Handle incoming requests (tool calls from sandbox)
        if "method" in msg:
            if msg["method"] == "tool_call":
                self._handle_tool_call(msg)
                continue

        if self._handle_out_of_band_message(msg, "during execution"):
            skipped += 1
            continue

        # Handle success response
        if "result" in msg:
            if msg.get("id") != execute_request_id:
                self._raise_terminal_error(
                    f"Response ID mismatch: expected {execute_request_id}, got {msg.get('id')}"
                )
            result = msg["result"]
            if not isinstance(result, dict):
                self._raise_terminal_error(f"Malformed execution result: {msg}")
            self._sync_files()
            # Check for SUBMIT (encoded as success with "final" field)
            if "final" in result:
                return FinalOutput(result["final"])
            return result.get("output", None)

        # Handle error response
        if "error" in msg:
            if msg.get("id") != execute_request_id:
                self._raise_terminal_error(
                    f"Response ID mismatch: expected {execute_request_id}, got {msg.get('id')}"
                )
            error = msg["error"]
            if not isinstance(error, dict):
                self._raise_terminal_error(f"Malformed execution error: {msg}")
            error_code = error.get("code", JSONRPC_APP_ERRORS["Unknown"])
            error_message = error.get("message", "Unknown error")
            error_data = error.get("data", {})
            if not isinstance(error_data, dict):
                self._raise_terminal_error(f"Malformed execution error data: {msg}")
            error_type = error_data.get("type", "Error")

            if error_code == JSONRPC_APP_ERRORS["SyntaxError"]:
                raise SyntaxError(f"Invalid Python syntax. message: {error_message}")
            if error_code in JSONRPC_APP_ERRORS.values():
                raise CodeExecutionError(f"{error_type}: {error_data.get('args') or error_message}")
            self._raise_terminal_error(f"{error_type}: {error_data.get('args') or error_message}")

        # Unexpected message format - neither a recognized method nor a response
        self._raise_terminal_error(f"Unexpected message format from sandbox: {msg}")

    self._raise_terminal_error(f"Too many skipped lines ({skipped}) during execution")

@with_callbacks
def shutdown(self) -> None:
    session_was_active = not self._session_ended
    self._session_ended = True
    if self.deno_process and self.deno_process.poll() is None:
        if session_was_active:
            self.deno_process.stdin.write(_jsonrpc_notification("shutdown") + "\n")
            self.deno_process.stdin.flush()
            self.deno_process.stdin.close()
        else:
            self.deno_process.terminate()
        self.deno_process.wait()
    self.deno_process = None
    self._owner_thread = None

@with_callbacks
def start(self) -> None:
    """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 while the session is active. A stopped or shut-down session
    cannot be restarted because its Python state cannot be reconstructed.
    """
    if self.deno_process is None:
        self._check_session_active()
        self._spawn_process()
    else:
        self._ensure_deno_process()