Execute Python code string. Uses sandbox mode or local mode based on use_sandbox setting. Args: code: Python code to execute. skill_id: Identifier of the skill being executed. input_spec: Input specification. Returns:
(
self,
code: str,
skill_id: str = 'unknown',
input_spec: ExecutionInput = None)
| 950 | return output |
| 951 | |
| 952 | async def execute_python_code( |
| 953 | self, |
| 954 | code: str, |
| 955 | skill_id: str = 'unknown', |
| 956 | input_spec: ExecutionInput = None) -> ExecutionOutput: |
| 957 | """ |
| 958 | Execute Python code string. |
| 959 | |
| 960 | Uses sandbox mode or local mode based on use_sandbox setting. |
| 961 | |
| 962 | Args: |
| 963 | code: Python code to execute. |
| 964 | skill_id: Identifier of the skill being executed. |
| 965 | input_spec: Input specification. |
| 966 | |
| 967 | Returns: |
| 968 | ExecutionOutput with results. |
| 969 | """ |
| 970 | input_spec = input_spec or ExecutionInput() |
| 971 | |
| 972 | record = self._create_record( |
| 973 | skill_id=skill_id, |
| 974 | executor_type=ExecutorType.PYTHON_CODE, |
| 975 | input_spec=input_spec, |
| 976 | script_path='<inline>') |
| 977 | |
| 978 | record.start_time = datetime.now() |
| 979 | record.status = ExecutionStatus.RUNNING |
| 980 | |
| 981 | try: |
| 982 | # Security check (stricter for local mode) |
| 983 | is_safe, reason = self._security_check( |
| 984 | code, is_local=not self.use_sandbox) |
| 985 | if not is_safe: |
| 986 | record.status = ExecutionStatus.SECURITY_BLOCKED |
| 987 | record.error_message = reason |
| 988 | output = ExecutionOutput( |
| 989 | stderr=f'Security check failed: {reason}', exit_code=-1) |
| 990 | record.end_time = datetime.now() |
| 991 | record.output_spec = output |
| 992 | self.spec.add_record(record) |
| 993 | return output |
| 994 | |
| 995 | start_time = datetime.now() |
| 996 | |
| 997 | if self.use_sandbox: |
| 998 | # Sandbox mode |
| 999 | env_setup = self._generate_env_setup(input_spec, {}) |
| 1000 | full_code = env_setup + '\n' + code |
| 1001 | |
| 1002 | results = await self._execute_in_sandbox( |
| 1003 | python_code=full_code, |
| 1004 | requirements=input_spec.requirements) |
| 1005 | stdout, stderr, exit_code = self._parse_sandbox_result(results) |
| 1006 | else: |
| 1007 | # Local mode |
| 1008 | stdout, stderr, exit_code = await self._local_execute_python_code( |
| 1009 | code, input_spec) |