Execute a Python script file. Uses sandbox mode or local mode based on use_sandbox setting. Args: script_path: Path to the Python script. skill_id: Identifier of the skill being executed. input_spec: Input specification. Returns
(
self,
script_path: Union[str, Path],
skill_id: str = 'unknown',
input_spec: ExecutionInput = None)
| 862 | requirements=requirements) |
| 863 | |
| 864 | async def execute_python_script( |
| 865 | self, |
| 866 | script_path: Union[str, Path], |
| 867 | skill_id: str = 'unknown', |
| 868 | input_spec: ExecutionInput = None) -> ExecutionOutput: |
| 869 | """ |
| 870 | Execute a Python script file. |
| 871 | |
| 872 | Uses sandbox mode or local mode based on use_sandbox setting. |
| 873 | |
| 874 | Args: |
| 875 | script_path: Path to the Python script. |
| 876 | skill_id: Identifier of the skill being executed. |
| 877 | input_spec: Input specification. |
| 878 | |
| 879 | Returns: |
| 880 | ExecutionOutput with results. |
| 881 | """ |
| 882 | input_spec = input_spec or ExecutionInput() |
| 883 | script_path = Path(script_path) |
| 884 | |
| 885 | record = self._create_record( |
| 886 | skill_id=skill_id, |
| 887 | executor_type=ExecutorType.PYTHON_SCRIPT, |
| 888 | input_spec=input_spec, |
| 889 | script_path=str(script_path)) |
| 890 | |
| 891 | record.start_time = datetime.now() |
| 892 | record.status = ExecutionStatus.RUNNING |
| 893 | |
| 894 | try: |
| 895 | # Read script content |
| 896 | with open(script_path, 'r', encoding='utf-8') as f: |
| 897 | code = f.read() |
| 898 | |
| 899 | # Security check (stricter for local mode) |
| 900 | is_safe, reason = self._security_check( |
| 901 | code, is_local=not self.use_sandbox) |
| 902 | if not is_safe: |
| 903 | record.status = ExecutionStatus.SECURITY_BLOCKED |
| 904 | record.error_message = reason |
| 905 | output = ExecutionOutput( |
| 906 | stderr=f'Security check failed: {reason}', exit_code=-1) |
| 907 | record.end_time = datetime.now() |
| 908 | record.output_spec = output |
| 909 | self.spec.add_record(record) |
| 910 | return output |
| 911 | |
| 912 | start_time = datetime.now() |
| 913 | |
| 914 | if self.use_sandbox: |
| 915 | # Sandbox mode: inject environment and execute |
| 916 | env_setup = self._generate_env_setup(input_spec, {}) |
| 917 | full_code = env_setup + '\n' + code |
| 918 | |
| 919 | results = await self._execute_in_sandbox( |
| 920 | python_code=full_code, |
| 921 | requirements=input_spec.requirements) |
no test coverage detected