Skip to content

Executors API Reference

Bases: Protocol

Protocol for objects that execute file-based skill scripts.

Implement this to run skill scripts somewhere other than a local subprocess — a container sandbox, a remote worker, or an in-process debugger. Pass the instance as script_executor to SkillsDirectory or discover_skills.

Example 
from typing import Any

from pydantic_ai_skills import SkillScript, SkillsDirectory


class EchoExecutor:
    async def run(
        self,
        script: SkillScript,
        args: dict[str, Any] | None = None,
        ctx: Any | None = None,
    ) -> Any:
        return f'Would run {script.uri} with {args}'


directory = SkillsDirectory(path='./skills', script_executor=EchoExecutor())
Source code in pydantic_ai_skills/executors.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@runtime_checkable
class SkillScriptExecutor(Protocol):
    """Protocol for objects that execute file-based skill scripts.

    Implement this to run skill scripts somewhere other than a local
    subprocess — a container sandbox, a remote worker, or an in-process
    debugger. Pass the instance as ``script_executor`` to ``SkillsDirectory``
    or ``discover_skills``.

    Example:
        ```python
        from typing import Any

        from pydantic_ai_skills import SkillScript, SkillsDirectory


        class EchoExecutor:
            async def run(
                self,
                script: SkillScript,
                args: dict[str, Any] | None = None,
                ctx: Any | None = None,
            ) -> Any:
                return f'Would run {script.uri} with {args}'


        directory = SkillsDirectory(path='./skills', script_executor=EchoExecutor())
        ```
    """

    async def run(
        self,
        script: SkillScript,
        args: dict[str, Any] | None = None,
        ctx: Any | None = None,
    ) -> Any:
        """Run a skill script.

        Args:
            script: The script to run. For file-based scripts, ``script.uri``
                holds the path to the script file.
            args: Named arguments for the script, or None.
            ctx: Optional run context, forwarded from the agent run.

        Returns:
            The script output. Executors used with ``run_skill_script``
            should return a string.
        """
        ...

run async 

run(script: SkillScript, args: dict[str, Any] | None = None, ctx: Any | None = None) -> Any

Run a skill script.

Parameters:

Name Type Description Default
script SkillScript

The script to run. For file-based scripts, script.uri holds the path to the script file.

 required
args dict[str, Any] | None

Named arguments for the script, or None.

 None
ctx Any | None

Optional run context, forwarded from the agent run.

 None

Returns:

Type Description
Any

The script output. Executors used with run_skill_script
Any

should return a string.
Source code in pydantic_ai_skills/executors.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def run(
    self,
    script: SkillScript,
    args: dict[str, Any] | None = None,
    ctx: Any | None = None,
) -> Any:
    """Run a skill script.

    Args:
        script: The script to run. For file-based scripts, ``script.uri``
            holds the path to the script file.
        args: Named arguments for the script, or None.
        ctx: Optional run context, forwarded from the agent run.

    Returns:
        The script output. Executors used with ``run_skill_script``
        should return a string.
    """
    ...

Bases: SkillScriptExecutor

Execute skill scripts using local subprocesses.

Executes file-based scripts as subprocesses with args passed as command-line named arguments. Dictionary keys are used exactly as provided (e.g., {"max-papers": 5} becomes --max-papers 5). A shebang line is used first when present and resolvable, then suffix-based fallback is used for compatibility. Other files are executed directly. Uses anyio.open_process with custom output collection and timeout handling for async-compatible subprocess execution.

Note

All scripts must accept named arguments. Positional arguments are not supported.

Attributes:

Name Type Description
timeout

Execution timeout in seconds.
Source code in pydantic_ai_skills/local.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
class LocalSkillScriptExecutor(SkillScriptExecutor):
    """Execute skill scripts using local subprocesses.

    Executes file-based scripts as subprocesses with args passed as command-line named arguments.
    Dictionary keys are used exactly as provided (e.g., {"max-papers": 5} becomes --max-papers 5).
    A shebang line is used first when present and resolvable, then suffix-based fallback
    is used for compatibility. Other files are executed directly.
    Uses anyio.open_process with custom output collection and timeout handling for
    async-compatible subprocess execution.

    Note:
        All scripts must accept named arguments. Positional arguments are not supported.

    Attributes:
        timeout: Execution timeout in seconds.
    """

    _SHELL_INTERPRETERS: dict[str, list[str]] = {
        '.sh': ['sh'],
        '.bash': ['bash'],
        '.zsh': ['zsh'],
        '.fish': ['fish'],
        '.bat': ['cmd', '/c'],
        '.cmd': ['cmd', '/c'],
    }

    def __init__(
        self,
        python_executable: str | Path | None = None,
        timeout: int = 30,
        env_vars: dict[str, str] | None = None,
        context_env_vars_extractor: ContextEnvVarsExtractor | None = None,
    ) -> None:
        """Initialize the local script executor.

        Args:
            python_executable: Path to Python executable. If None, uses sys.executable.
            timeout: Execution timeout in seconds (default: 30).
            env_vars: Optional static environment variables merged into every script
                subprocess environment.
            context_env_vars_extractor: Optional callable that extracts context-specific
                environment variables from run context. When omitted, no context
                environment variables are forwarded.
        """
        self._python_executable = str(python_executable) if python_executable else sys.executable
        self.timeout = timeout
        self._env_vars = self._coerce_mapping_to_env_vars(env_vars)
        self._context_env_vars_extractor = context_env_vars_extractor

    @staticmethod
    def _coerce_mapping_to_env_vars(value: Any) -> dict[str, str]:
        """Convert mapping-like values into subprocess environment variables."""
        if not isinstance(value, Mapping):
            return {}

        mapping = cast(Mapping[Any, Any], value)
        env_vars: dict[str, str] = {}
        for key, item in mapping.items():
            key_str = str(key)
            if not key_str:
                continue
            if item is None:
                continue
            env_vars[key_str] = str(item)

        return env_vars

    def _build_context_env_vars(self, ctx: Any | None) -> dict[str, str]:
        """Build context-provided environment variables from run context."""
        if ctx is None or self._context_env_vars_extractor is None:
            return {}

        extracted = self._context_env_vars_extractor(ctx)
        return self._coerce_mapping_to_env_vars(extracted)

    def _build_process_env(self, ctx: Any | None) -> dict[str, str] | None:
        """Build subprocess environment using static and context env vars."""
        context_env_vars = self._build_context_env_vars(ctx)
        if not self._env_vars and not context_env_vars:
            return None

        process_env = os.environ.copy()
        process_env.update(self._env_vars)
        process_env.update(context_env_vars)
        return process_env

    @staticmethod
    def _resolve_interpreter(interpreter: str) -> str | None:
        """Resolve a shebang interpreter to an executable path."""
        if Path(interpreter).is_absolute():
            path = Path(interpreter)
            return str(path) if path.exists() else None
        return shutil.which(interpreter)

    def _extract_shebang_command(self, script_path: Path) -> list[str] | None:
        """Return an interpreter command from shebang if present and resolvable."""
        try:
            with script_path.open('rb') as handle:
                first_line = handle.readline()
        except OSError:
            return None

        if not first_line.startswith(b'#!'):
            return None

        shebang = first_line[2:].decode('utf-8', errors='ignore').strip()
        if not shebang:
            return None

        parts = shlex.split(shebang)
        if not parts:
            return None

        if Path(parts[0]).name == 'env':
            idx = 1
            while idx < len(parts) and parts[idx].startswith('-'):
                idx += 1
            if idx >= len(parts):
                return None
            interpreter = parts[idx]
            interpreter_args = parts[idx + 1 :]
        else:
            interpreter = parts[0]
            interpreter_args = parts[1:]

        resolved = self._resolve_interpreter(interpreter)
        if not resolved:
            return None

        return [resolved, *interpreter_args]

    def _build_command(self, script_path: Path) -> list[str]:
        """Build subprocess command using shebang-first dispatch with compatibility fallback."""
        shebang_command = self._extract_shebang_command(script_path)
        if shebang_command:
            return [*shebang_command, str(script_path)]

        suffix = script_path.suffix.lower()
        if suffix == '.py':
            return [self._python_executable, str(script_path)]
        if suffix == '.ps1':
            powershell = shutil.which('pwsh') or shutil.which('powershell')
            if powershell:
                return [powershell, '-File', str(script_path)]
        if suffix in self._SHELL_INTERPRETERS:
            return [*self._SHELL_INTERPRETERS[suffix], str(script_path)]
        return [str(script_path)]

    @staticmethod
    def _build_args(cmd: list[str], args: dict[str, Any]) -> None:
        """Append named arguments to cmd in-place."""
        for key, value in args.items():
            if isinstance(value, bool):
                if value:
                    cmd.append(f'--{key}')
            elif isinstance(value, list):
                for item in cast(list[Any], value):
                    cmd.append(f'--{key}')
                    cmd.append(str(item))
            elif value is not None:
                cmd.append(f'--{key}')
                cmd.append(str(value))

    @staticmethod
    def _format_output(stdout_chunks: list[bytes], stderr_chunks: list[bytes], return_code: int) -> str:
        """Decode and combine stdout, stderr, and exit code into a single string."""
        output = b''.join(stdout_chunks).decode('utf-8', errors='replace')
        stderr_output = b''.join(stderr_chunks).decode('utf-8', errors='replace')
        if stderr_output:
            output += f'\n\nStderr:\n{stderr_output}'
        if return_code != 0:
            output += f'\n\nScript exited with code {return_code}'
        return output.strip() or '(no output)'

    async def _drain_stream(self, stream: anyio.abc.ByteReceiveStream | None, chunks: list[bytes]) -> None:
        """Drain a process output stream into chunks until EOF."""
        if stream is None:
            return

        while True:
            try:
                chunk = await stream.receive()
            except anyio.EndOfStream:
                break

            if chunk == b'':
                break

            chunks.append(chunk)

    async def _collect_output(
        self,
        process: anyio.abc.Process,
        stdout_chunks: list[bytes],
        stderr_chunks: list[bytes],
    ) -> int:
        """Read stdout/stderr concurrently, then wait for the process to exit."""
        async with anyio.create_task_group() as io_tg:
            io_tg.start_soon(self._drain_stream, process.stdout, stdout_chunks)
            io_tg.start_soon(self._drain_stream, process.stderr, stderr_chunks)

        return await process.wait()

    @staticmethod
    def _kill_process(process: anyio.abc.Process, use_process_group: bool) -> None:
        """Terminate a process and its process group when possible."""
        if use_process_group:
            try:
                # Validate PID before calling os.getpgid
                if process.pid and process.pid > 0:
                    pgid = os.getpgid(process.pid)
                    if pgid > 0:
                        os.killpg(pgid, signal.SIGKILL)
                        return
            except (OSError, ValueError):
                # Process no longer exists or group doesn't exist
                pass
        try:
            process.kill()
        except OSError:
            # Process already terminated
            pass

    async def _run_with_timeout(
        self,
        process: anyio.abc.Process,
        stdout_chunks: list[bytes],
        stderr_chunks: list[bytes],
        use_process_group: bool,
    ) -> tuple[int, bool]:
        """Collect process output while enforcing timeout."""
        return_code = 0
        timed_out = False

        def _kill() -> None:
            nonlocal timed_out
            timed_out = True
            self._kill_process(process, use_process_group)

        async with anyio.create_task_group() as tg:

            async def _kill_after_timeout() -> None:
                await anyio.sleep(self.timeout)
                _kill()

            async def _run() -> None:
                nonlocal return_code
                return_code = await self._collect_output(process, stdout_chunks, stderr_chunks)
                tg.cancel_scope.cancel()

            tg.start_soon(_kill_after_timeout)
            tg.start_soon(_run)

        return return_code, timed_out

    async def _start_process(
        self,
        cmd: list[str],
        cwd: str,
        use_process_group: bool,
        script_name: str,
        env: dict[str, str] | None = None,
    ) -> anyio.abc.Process:
        """Start script subprocess and normalize startup errors."""
        try:
            return await anyio.open_process(
                cmd,
                stdin=_subprocess.DEVNULL,
                stdout=_subprocess.PIPE,
                stderr=_subprocess.PIPE,
                cwd=cwd,
                env=env,
                start_new_session=use_process_group,
            )
        except OSError as e:
            raise RuntimeError(f"Failed to execute script '{script_name}': {e}") from e

    async def run(
        self,
        script: SkillScript,
        args: dict[str, Any] | None = None,
        ctx: Any | None = None,
    ) -> Any:
        """Run a skill script locally using subprocess.

        Args:
            script: The script to run.
            args: Named arguments as a dictionary.
                Boolean True emits flag only, False/None omits it,
                lists repeat the flag for each item, other types convert to string.
            ctx: Optional run context used to merge context-provided env_vars into
                the subprocess environment.

        Returns:
            Combined stdout and stderr output.

        Raises:
            ValueError: If the script has no URI configured.
            RuntimeError: If execution fails to start.
            TimeoutError: If execution exceeds the configured timeout.
        """
        if script.uri is None:
            raise ValueError(f"Script '{script.name}' has no URI for subprocess execution")

        script_path = Path(script.uri)
        cmd = self._build_command(script_path)
        if args:
            self._build_args(cmd, args)

        process_env = self._build_process_env(ctx)

        cwd = str(script_path.parent)
        use_process_group = sys.platform != 'win32'
        process = await self._start_process(
            cmd,
            cwd,
            use_process_group,
            script.name,
            env=process_env,
        )

        stdout_chunks: list[bytes] = []
        stderr_chunks: list[bytes] = []
        return_code = 0
        timed_out = False

        try:
            return_code, timed_out = await self._run_with_timeout(
                process=process,
                stdout_chunks=stdout_chunks,
                stderr_chunks=stderr_chunks,
                use_process_group=use_process_group,
            )
        finally:
            try:
                await process.aclose()
            except OSError:
                pass

        if timed_out:
            raise TimeoutError(f"Script '{script.name}' timed out after {self.timeout} seconds")

        return self._format_output(stdout_chunks, stderr_chunks, return_code)

run async 

run(script: SkillScript, args: dict[str, Any] | None = None, ctx: Any | None = None) -> Any

Run a skill script locally using subprocess.

Parameters:

Name Type Description Default
script SkillScript

The script to run.

 required
args dict[str, Any] | None

Named arguments as a dictionary. Boolean True emits flag only, False/None omits it, lists repeat the flag for each item, other types convert to string.

 None
ctx Any | None

Optional run context used to merge context-provided env_vars into the subprocess environment.

 None

Returns:

Type Description
Any

Combined stdout and stderr output.

Raises:

Type Description
ValueError

If the script has no URI configured.
RuntimeError

If execution fails to start.
TimeoutError

If execution exceeds the configured timeout.
Source code in pydantic_ai_skills/local.py 
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
async def run(
    self,
    script: SkillScript,
    args: dict[str, Any] | None = None,
    ctx: Any | None = None,
) -> Any:
    """Run a skill script locally using subprocess.

    Args:
        script: The script to run.
        args: Named arguments as a dictionary.
            Boolean True emits flag only, False/None omits it,
            lists repeat the flag for each item, other types convert to string.
        ctx: Optional run context used to merge context-provided env_vars into
            the subprocess environment.

    Returns:
        Combined stdout and stderr output.

    Raises:
        ValueError: If the script has no URI configured.
        RuntimeError: If execution fails to start.
        TimeoutError: If execution exceeds the configured timeout.
    """
    if script.uri is None:
        raise ValueError(f"Script '{script.name}' has no URI for subprocess execution")

    script_path = Path(script.uri)
    cmd = self._build_command(script_path)
    if args:
        self._build_args(cmd, args)

    process_env = self._build_process_env(ctx)

    cwd = str(script_path.parent)
    use_process_group = sys.platform != 'win32'
    process = await self._start_process(
        cmd,
        cwd,
        use_process_group,
        script.name,
        env=process_env,
    )

    stdout_chunks: list[bytes] = []
    stderr_chunks: list[bytes] = []
    return_code = 0
    timed_out = False

    try:
        return_code, timed_out = await self._run_with_timeout(
            process=process,
            stdout_chunks=stdout_chunks,
            stderr_chunks=stderr_chunks,
            use_process_group=use_process_group,
        )
    finally:
        try:
            await process.aclose()
        except OSError:
            pass

    if timed_out:
        raise TimeoutError(f"Script '{script.name}' timed out after {self.timeout} seconds")

    return self._format_output(stdout_chunks, stderr_chunks, return_code)

Bases: SkillScriptExecutor

Wraps a callable in a script executor interface.

Allows users to provide custom execution logic for file-based scripts instead of using subprocess execution. Useful for remote execution, sandboxed execution, or other custom scenarios.

Example 
from pydantic_ai_skills import CallableSkillScriptExecutor, SkillsDirectory

async def my_executor(script, args=None):
    # Custom execution logic - script.uri contains the file path
    return f"Executed {script.name} at {script.uri} with {args}"

executor = CallableSkillScriptExecutor(func=my_executor)
directory = SkillsDirectory(path="./skills", script_executor=executor)
Source code in pydantic_ai_skills/local.py 
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
class CallableSkillScriptExecutor(SkillScriptExecutor):
    """Wraps a callable in a script executor interface.

    Allows users to provide custom execution logic for file-based scripts
    instead of using subprocess execution. Useful for remote execution, sandboxed
    execution, or other custom scenarios.

    Example:
        ```python
        from pydantic_ai_skills import CallableSkillScriptExecutor, SkillsDirectory

        async def my_executor(script, args=None):
            # Custom execution logic - script.uri contains the file path
            return f"Executed {script.name} at {script.uri} with {args}"

        executor = CallableSkillScriptExecutor(func=my_executor)
        directory = SkillsDirectory(path="./skills", script_executor=executor)
        ```
    """

    def __init__(self, func: Callable[..., Any]) -> None:
        """Initialize the callable executor.

        Args:
            func: Callable that executes scripts. Can be sync or async.
                Should accept keyword arguments: script (SkillScript) and args (dict[str, Any] | None).
                It may also receive ctx when the callable declares a ctx keyword parameter or accepts
                arbitrary keyword arguments via **kwargs. Should return the script output as a string.
                The script's uri attribute contains the file path.
        """
        self._func = func
        self._is_async = is_async_callable(func)
        self._accepts_ctx = self._callable_accepts_keyword(func, 'ctx')

    @staticmethod
    def _callable_accepts_keyword(func: Callable[..., Any], keyword: str) -> bool:
        """Return True if callable accepts the given keyword argument."""
        try:
            signature = inspect.signature(func)
        except (TypeError, ValueError):
            return False

        for parameter in signature.parameters.values():
            if parameter.kind == inspect.Parameter.VAR_KEYWORD:
                return True

        parameter = signature.parameters.get(keyword)
        if parameter is None:
            return False

        return parameter.kind != inspect.Parameter.POSITIONAL_ONLY

    async def run(
        self,
        script: SkillScript,
        args: dict[str, Any] | None = None,
        ctx: Any | None = None,
    ) -> Any:
        """Run using the wrapped callable.

        Args:
            script: The script to run.
            args: Named arguments as a dictionary.
            ctx: Optional run context passed to wrapped callable when supported.

        Returns:
            Script output (can be any type like str, dict, etc.).
        """
        kwargs: dict[str, Any] = {
            'script': script,
            'args': args,
        }
        if ctx is not None and self._accepts_ctx:
            kwargs['ctx'] = ctx

        if self._is_async:
            function = cast(Callable[..., Awaitable[Any]], self._func)
            return await function(**kwargs)
        else:
            return await run_in_executor(self._func, **kwargs)

run async 

run(script: SkillScript, args: dict[str, Any] | None = None, ctx: Any | None = None) -> Any

Run using the wrapped callable.

Parameters:

Name Type Description Default
script SkillScript

The script to run.

 required
args dict[str, Any] | None

Named arguments as a dictionary.

 None
ctx Any | None

Optional run context passed to wrapped callable when supported.

 None

Returns:

Type Description
Any

Script output (can be any type like str, dict, etc.).
Source code in pydantic_ai_skills/local.py 
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
async def run(
    self,
    script: SkillScript,
    args: dict[str, Any] | None = None,
    ctx: Any | None = None,
) -> Any:
    """Run using the wrapped callable.

    Args:
        script: The script to run.
        args: Named arguments as a dictionary.
        ctx: Optional run context passed to wrapped callable when supported.

    Returns:
        Script output (can be any type like str, dict, etc.).
    """
    kwargs: dict[str, Any] = {
        'script': script,
        'args': args,
    }
    if ctx is not None and self._accepts_ctx:
        kwargs['ctx'] = ctx

    if self._is_async:
        function = cast(Callable[..., Awaitable[Any]], self._func)
        return await function(**kwargs)
    else:
        return await run_in_executor(self._func, **kwargs)

Execute file-based skill scripts inside an OpenSandbox container.

Attributes:

Name Type Description
timeout

Per-script execution timeout in seconds.
Source code in pydantic_ai_skills/sandboxes/opensandbox.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
class OpenSandboxScriptExecutor:
    """Execute file-based skill scripts inside an OpenSandbox container.

    Attributes:
        timeout: Per-script execution timeout in seconds.
    """

    def __init__(
        self,
        image: str = 'opensandbox/code-interpreter:v1.1.0',
        *,
        timeout: int = 30,
        workdir: str = '/workspace/skills',
        env_vars: dict[str, str] | None = None,
        reuse_sandbox: bool = False,
        sandbox_timeout: timedelta = timedelta(minutes=10),
    ) -> None:
        """Initialize the OpenSandbox executor.

        Args:
            image: Container image used for each sandbox.
            timeout: Per-script execution timeout in seconds.
            workdir: Directory inside the sandbox that the skill folder is staged into.
                Deliberately not under ``/tmp``: that is world-writable, so another
                process in the sandbox could tamper with a staged script between
                upload and execution. The directory is created if missing.
            env_vars: Environment variables exported to the script process.
            reuse_sandbox: Keep a single sandbox alive across runs instead of
                creating and killing one per run. Faster, but runs share state.
            sandbox_timeout: Lifetime of the sandbox itself, passed to ``Sandbox.create``.
        """
        self.timeout = timeout
        self._image = image
        self._workdir = workdir.rstrip('/') or _DEFAULT_WORKDIR
        self._env_vars = dict(env_vars or {})
        self._reuse_sandbox = reuse_sandbox
        self._sandbox_timeout = sandbox_timeout
        self._sandbox: Sandbox | None = None
        self._sandbox_deadline: float = 0.0
        self._staged_paths: set[str] = set()
        self._staged_dirs: set[str] = set()
        self._staged_root: Path | None = None
        self._staged_fingerprint: str | None = None
        # Serializes runs that share one sandbox; see run().
        self._reuse_lock = anyio.Lock()
        # Reused for its host-independent argument marshalling and output formatting.
        self._formatter = LocalSkillScriptExecutor()

    async def _get_sandbox(self) -> Sandbox:
        """Return the sandbox to run in, creating one when needed.

        A reused sandbox is replaced once its server-side lifetime is close to
        expiring. ``Sandbox.create`` fixes that lifetime, so holding the handle
        past it would send every later run to an expired sandbox. The deadline
        leaves one script timeout of headroom so a run started now can finish.
        """
        if self._reuse_sandbox and self._sandbox is not None:
            if time.monotonic() < self._sandbox_deadline:
                return self._sandbox
            # aclose() also clears the staging record: the replacement starts empty.
            await self.aclose()

        sandbox_cls = _require_opensandbox()
        sandbox: Sandbox = await sandbox_cls.create(
            self._image,
            env=self._env_vars or None,
            timeout=self._sandbox_timeout,
        )
        if self._reuse_sandbox:
            self._sandbox = sandbox
            self._sandbox_deadline = time.monotonic() + self._sandbox_timeout.total_seconds() - self.timeout
        return sandbox

    async def _stage_skill_folder(self, sandbox: Sandbox, skill_root: Path) -> None:
        """Upload the skill folder into the sandbox workdir.

        A reused sandbox keeps whatever earlier runs wrote, so restaging has to
        remove files that no longer exist in the source skill. Without that, a
        resource deleted or renamed between runs stays readable and the script
        goes on using stale data. When nothing changed, staging is skipped.
        """
        from opensandbox.models import WriteEntry

        entries, source_dirs, fingerprint = _stage_snapshot(skill_root)
        # Keyed on the root as well: two skills can share relative paths and
        # contents, and fingerprint alone would then run skill B against skill A's
        # staged files.
        if self._reuse_sandbox and (skill_root, fingerprint) == (self._staged_root, self._staged_fingerprint):
            return

        paths = {f'{self._workdir}/{entry.relative}' for entry in entries}
        # Every ancestor, not just the immediate parent: creating resources/a/b also
        # leaves resources/a behind, and an untracked ancestor would survive pruning
        # and block a later skill that needs a file at that path.
        # Every source directory, so a skill's empty scratch/ exists too.
        directories = {self._workdir} | {f'{self._workdir}/{name}' for name in source_dirs}
        for entry in entries:
            parent = PurePosixPath(entry.relative).parent
            while parent != PurePosixPath('.'):
                directories.add(f'{self._workdir}/{parent}')
                parent = parent.parent

        stale_files = sorted(self._staged_paths - paths)
        if stale_files:
            await sandbox.files.delete_files(stale_files)

        # Directories too: a path that was a directory in the previous skill and is
        # a file in this one would otherwise block the write. Deepest first so
        # children go before their parents.
        stale_dirs = sorted(self._staged_dirs - directories, key=lambda path: path.count('/'), reverse=True)
        if stale_dirs:
            await sandbox.files.delete_directories(stale_dirs)

        await sandbox.files.create_directories([WriteEntry(path=path) for path in sorted(directories)])

        if entries:
            await sandbox.files.write_files(
                [
                    WriteEntry(
                        path=f'{self._workdir}/{entry.relative}',
                        data=entry.data,
                        mode=0o755 if entry.executable else 0o644,
                    )
                    for entry in entries
                ]
            )

        if self._reuse_sandbox:
            self._staged_paths = paths
            self._staged_dirs = directories
            self._staged_root = skill_root
            self._staged_fingerprint = fingerprint

    def _build_command(self, script_path: Path, remote_path: str, suffix: str, args: dict[str, Any] | None) -> str:
        """Build the shell command line executed inside the sandbox.

        A shebang wins over the suffix fallback, matching
        [`LocalSkillScriptExecutor`][pydantic_ai_skills.LocalSkillScriptExecutor];
        otherwise a ``#!/bin/bash`` script using bash-only syntax would run under
        ``sh`` here but bash locally.
        """
        interpreter = _shebang_command(script_path) or _SANDBOX_INTERPRETERS.get(suffix)
        cmd = [*interpreter, remote_path] if interpreter else [remote_path]

        if args:
            # Reuse the built-in bool/list/None marshalling rules.
            self._formatter._build_args(cmd, args)

        return shlex.join(cmd)

    async def run(
        self,
        script: SkillScript,
        args: dict[str, Any] | None = None,
        ctx: Any | None = None,
    ) -> Any:
        """Run a skill script inside an OpenSandbox container.

        Args:
            script: The script to run; ``script.uri`` must point at a local file.
            args: Named arguments, marshalled with the same rules as
                [`LocalSkillScriptExecutor`][pydantic_ai_skills.LocalSkillScriptExecutor].
            ctx: Unused; accepted for protocol compatibility.

        Returns:
            Combined stdout and stderr, formatted like local execution.

        Raises:
            ValueError: If the script has no URI configured.
        """
        del ctx  # Required by the SkillScriptExecutor protocol; unused by this backend.

        if script.uri is None:
            raise ValueError(f"Script '{script.name}' has no URI for sandbox execution")

        script_path = Path(script.uri).resolve()
        skill_root = skill_root_for(script)
        remote_path = f'{self._workdir}/{script_path.relative_to(skill_root).as_posix()}'
        # cwd is the script's own directory, matching LocalSkillScriptExecutor.
        working_directory = str(PurePosixPath(remote_path).parent)
        command = self._build_command(script_path, remote_path, script_path.suffix.lower(), args)

        if not self._reuse_sandbox:
            return await self._execute(skill_root, command, working_directory)

        # One sandbox serving concurrent runs has to serialize them: two first
        # runs would otherwise each create a container and leak one, and both
        # would stage over each other's files in the shared workdir.
        async with self._reuse_lock:
            return await self._execute(skill_root, command, working_directory)

    async def _execute(self, skill_root: Path, command: str, working_directory: str) -> Any:
        """Provision a sandbox, stage the skill, run the command, and format the output."""
        # _get_sandbox raises the ImportError naming the extra, so import the SDK
        # models only once a sandbox exists.
        sandbox = await self._get_sandbox()
        from opensandbox.models.execd import RunCommandOpts

        try:
            await self._stage_skill_folder(sandbox, skill_root)
            execution = await sandbox.commands.run(
                command,
                opts=RunCommandOpts(
                    working_directory=working_directory,
                    timeout=timedelta(seconds=self.timeout),
                    envs=self._env_vars or None,
                ),
            )
        finally:
            if not self._reuse_sandbox:
                # Shielded: a cancelled scope would cancel this await too, leaking
                # the container until its server-side lifetime expires.
                with anyio.CancelScope(shield=True):
                    await sandbox.kill()

        stdout = ''.join(message.text for message in execution.logs.stdout)
        stderr = ''.join(message.text for message in execution.logs.stderr)
        return self._formatter._format_output([stdout.encode()], [stderr.encode()], execution.exit_code or 0)

    async def aclose(self) -> None:
        """Kill the reused sandbox, if one is alive."""
        if self._sandbox is not None:
            await self._sandbox.kill()
            self._sandbox = None
            self._sandbox_deadline = 0.0
            self._staged_paths = set()
            self._staged_dirs = set()
            self._staged_root = None
            self._staged_fingerprint = None

run async 

run(script: SkillScript, args: dict[str, Any] | None = None, ctx: Any | None = None) -> Any

Run a skill script inside an OpenSandbox container.

Parameters:

Name Type Description Default
script SkillScript

The script to run; script.uri must point at a local file.

 required
args dict[str, Any] | None

Named arguments, marshalled with the same rules as LocalSkillScriptExecutor.

 None
ctx Any | None

Unused; accepted for protocol compatibility.

 None

Returns:

Type Description
Any

Combined stdout and stderr, formatted like local execution.

Raises:

Type Description
ValueError

If the script has no URI configured.
Source code in pydantic_ai_skills/sandboxes/opensandbox.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
async def run(
    self,
    script: SkillScript,
    args: dict[str, Any] | None = None,
    ctx: Any | None = None,
) -> Any:
    """Run a skill script inside an OpenSandbox container.

    Args:
        script: The script to run; ``script.uri`` must point at a local file.
        args: Named arguments, marshalled with the same rules as
            [`LocalSkillScriptExecutor`][pydantic_ai_skills.LocalSkillScriptExecutor].
        ctx: Unused; accepted for protocol compatibility.

    Returns:
        Combined stdout and stderr, formatted like local execution.

    Raises:
        ValueError: If the script has no URI configured.
    """
    del ctx  # Required by the SkillScriptExecutor protocol; unused by this backend.

    if script.uri is None:
        raise ValueError(f"Script '{script.name}' has no URI for sandbox execution")

    script_path = Path(script.uri).resolve()
    skill_root = skill_root_for(script)
    remote_path = f'{self._workdir}/{script_path.relative_to(skill_root).as_posix()}'
    # cwd is the script's own directory, matching LocalSkillScriptExecutor.
    working_directory = str(PurePosixPath(remote_path).parent)
    command = self._build_command(script_path, remote_path, script_path.suffix.lower(), args)

    if not self._reuse_sandbox:
        return await self._execute(skill_root, command, working_directory)

    # One sandbox serving concurrent runs has to serialize them: two first
    # runs would otherwise each create a container and leak one, and both
    # would stage over each other's files in the shared workdir.
    async with self._reuse_lock:
        return await self._execute(skill_root, command, working_directory)

aclose async 

aclose() -> None

Kill the reused sandbox, if one is alive.

Source code in pydantic_ai_skills/sandboxes/opensandbox.py 
309
310
311
312
313
314
315
316
317
318
async def aclose(self) -> None:
    """Kill the reused sandbox, if one is alive."""
    if self._sandbox is not None:
        await self._sandbox.kill()
        self._sandbox = None
        self._sandbox_deadline = 0.0
        self._staged_paths = set()
        self._staged_dirs = set()
        self._staged_root = None
        self._staged_fingerprint = None

Execute file-based skill scripts inside a LocalSandbox virtual filesystem.

LocalSandbox has no CPython binary on PATH, so there are two paths: shell scripts run through abash, and .py scripts run through aexecute_python (Pyodide) with an injected sys.argv and runpy.run_path(..., run_name='__main__'), so argparse and if __name__ == '__main__' behave normally.

Pyodide ships a subset of the ecosystem and has no sockets or subprocesses, so scripts needing third-party wheels or network access will fail here.

Attributes:

Name Type Description
workdir

Directory inside the sandbox that the skill folder is staged into.
Source code in pydantic_ai_skills/sandboxes/localsandbox.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
class LocalSandboxScriptExecutor:
    """Execute file-based skill scripts inside a LocalSandbox virtual filesystem.

    LocalSandbox has no CPython binary on ``PATH``, so there are two paths:
    shell scripts run through ``abash``, and ``.py`` scripts run through
    ``aexecute_python`` (Pyodide) with an injected ``sys.argv`` and
    ``runpy.run_path(..., run_name='__main__')``, so ``argparse`` and
    ``if __name__ == '__main__'`` behave normally.

    Pyodide ships a subset of the ecosystem and has no sockets or subprocesses,
    so scripts needing third-party wheels or network access will fail here.

    Attributes:
        workdir: Directory inside the sandbox that the skill folder is staged into.
    """

    def __init__(
        self,
        *,
        workdir: str = '/data/skill',
        preset: Any | None = None,
        preload_packages: list[str] | None = None,
        reuse_sandbox: bool = False,
    ) -> None:
        """Initialize the LocalSandbox executor.

        Args:
            workdir: Directory inside the sandbox that the skill folder is staged into.
            preset: Optional ``localsandbox.ExecutionPreset`` controlling resource
                limits. When None, the SDK default (``NORMAL``) is used.
            preload_packages: Pyodide packages to preload before running Python scripts.
            reuse_sandbox: Keep a single sandbox alive across runs instead of
                creating one per run. Faster, but runs share state.
        """
        self.workdir = workdir.rstrip('/') or '/data/skill'
        self._preset = preset
        self._preload_packages = preload_packages
        self._reuse_sandbox = reuse_sandbox
        self._sandbox: LocalSandbox | None = None
        self._staged_root: Path | None = None
        self._staged_fingerprint: str | None = None
        # Serializes runs that share one sandbox; see run().
        self._reuse_lock = anyio.Lock()
        # Reused for its host-independent argument marshalling and output formatting.
        self._formatter = LocalSkillScriptExecutor()

    def _get_sandbox(self, skill_root: Path) -> LocalSandbox:
        """Return the sandbox to run in, creating and staging one when needed.

        Staging happens at construction, so a reused sandbox is rebuilt whenever
        the skill changes — either a different skill (one executor instance
        serves every skill in a ``SkillsDirectory``) or edited files under the
        same root, which ``auto_reload`` and ``reload()`` both surface.
        """
        entries, directories, fingerprint = _stage_snapshot(skill_root)

        if self._reuse_sandbox and self._sandbox is not None:
            if self._staged_root == skill_root and self._staged_fingerprint == fingerprint:
                return self._sandbox
            self.close()

        sandbox_cls = _require_localsandbox()
        files: dict[str, str | bytes] = {f'{self.workdir}/{entry.relative}': entry.data for entry in entries}
        kwargs: dict[str, Any] = {'files': files, 'cwd': self.workdir}
        if self._preset is not None:
            kwargs['preset'] = self._preset

        sandbox: LocalSandbox = sandbox_cls(**kwargs)
        if directories:
            # The files mapping cannot express an empty directory, and a skill may
            # ship one for its script to write into.
            paths = ' '.join(shlex.quote(f'{self.workdir}/{name}') for name in directories)
            sandbox.bash(f'mkdir -p {paths}')

        if self._reuse_sandbox:
            self._sandbox = sandbox
            self._staged_root = skill_root
            self._staged_fingerprint = fingerprint
        return sandbox

    async def _run_python(
        self, sandbox: LocalSandbox, remote_path: str, cwd: str, args: dict[str, Any] | None
    ) -> tuple[str, str, int]:
        """Run a Python script through Pyodide with an injected argv."""
        argv: list[str] = [PurePosixPath(remote_path).name]
        if args:
            self._formatter._build_args(argv, args)

        code = _PYTHON_WRAPPER.format(
            # repr, not json.dumps: JSON escapes non-BMP characters as UTF-16
            # surrogate pairs, which become two lone surrogates in Python source.
            argv=repr(argv),
            script_path=remote_path,
            exit_file=_EXIT_CODE_FILE,
        )
        result = await sandbox.aexecute_python(
            code,
            cwd=cwd,
            preload_packages=self._preload_packages,
        )

        stderr = result.stderr or ''
        if result.error:
            # The wrapper never reached the exit-code file. Pyodide usually mirrors the
            # traceback into stderr already, so append rather than replace or duplicate.
            if result.error not in stderr:
                stderr = f'{stderr}\n{result.error}' if stderr else result.error
            return result.stdout or '', stderr, result.exit_code or 1

        try:
            exit_code = int(sandbox.read_file(_EXIT_CODE_FILE))
        except (OSError, ValueError):  # pragma: no cover - wrapper always writes it
            exit_code = result.exit_code or 0

        return result.stdout or '', stderr, exit_code

    async def _run_shell(
        self,
        sandbox: LocalSandbox,
        script_path: Path,
        remote_path: str,
        cwd: str,
        suffix: str,
        args: dict[str, Any] | None,
    ) -> tuple[str, str, int]:
        """Run a shell script through just-bash from the script's own directory."""
        from localsandbox import CommandError

        # A shebang wins over the suffix, matching LocalSkillScriptExecutor.
        shell = _shebang_shell(script_path) or _SHELL_INTERPRETERS[suffix]
        cmd = [*shell, remote_path]
        if args:
            self._formatter._build_args(cmd, args)

        # abash takes no cwd, so change directory as part of the command; without
        # this a script reading ../resources/data.json would resolve it differently
        # than under LocalSkillScriptExecutor.
        command = f'cd {shlex.quote(cwd)} && {shlex.join(cmd)}'

        try:
            result = await sandbox.abash(command)
        except CommandError as exc:
            # abash raises on non-zero exit; surface it like local execution does.
            return exc.stdout, exc.stderr, exc.exit_code

        return result.stdout, result.stderr, result.exit_code

    async def run(
        self,
        script: SkillScript,
        args: dict[str, Any] | None = None,
        ctx: Any | None = None,
    ) -> Any:
        """Run a skill script inside a LocalSandbox virtual filesystem.

        Args:
            script: The script to run; ``script.uri`` must point at a local file.
            args: Named arguments, marshalled with the same rules as
                [`LocalSkillScriptExecutor`][pydantic_ai_skills.LocalSkillScriptExecutor].
            ctx: Unused; accepted for protocol compatibility.

        Returns:
            Combined stdout and stderr, formatted like local execution.

        Raises:
            ValueError: If the script has no URI, or its type is unsupported here.
        """
        del ctx  # Required by the SkillScriptExecutor protocol; unused by this backend.

        if script.uri is None:
            raise ValueError(f"Script '{script.name}' has no URI for sandbox execution")

        script_path = Path(script.uri).resolve()
        skill_root = skill_root_for(script)
        suffix = script_path.suffix.lower()
        remote_path = f'{self.workdir}/{script_path.relative_to(skill_root).as_posix()}'
        # cwd is the script's own directory, matching LocalSkillScriptExecutor.
        cwd = str(PurePosixPath(remote_path).parent)

        # Validated before provisioning, so an unsupported script never starts a sandbox.
        if suffix != '.py' and suffix not in _SHELL_INTERPRETERS:
            raise ValueError(
                f"Script '{script.name}' has unsupported type '{suffix}' for LocalSandbox. "
                f'Supported: .py (Pyodide), {", ".join(sorted(_SHELL_INTERPRETERS))} (just-bash).'
            )

        if not self._reuse_sandbox:
            return await self._execute(skill_root, script_path, remote_path, cwd, suffix, args)

        # One sandbox serving concurrent runs has to serialize them: a second run
        # switching skills would otherwise close the sandbox the first is still
        # using, and both would share a filesystem mid-execution anyway.
        async with self._reuse_lock:
            return await self._execute(skill_root, script_path, remote_path, cwd, suffix, args)

    async def _execute(
        self,
        skill_root: Path,
        script_path: Path,
        remote_path: str,
        cwd: str,
        suffix: str,
        args: dict[str, Any] | None,
    ) -> Any:
        """Provision a sandbox, run the script in it, and format the output."""
        sandbox = self._get_sandbox(skill_root)
        try:
            if suffix == '.py':
                stdout, stderr, exit_code = await self._run_python(sandbox, remote_path, cwd, args)
            else:
                stdout, stderr, exit_code = await self._run_shell(sandbox, script_path, remote_path, cwd, suffix, args)
        finally:
            if not self._reuse_sandbox:
                sandbox.__exit__(None, None, None)

        return self._formatter._format_output([stdout.encode()], [stderr.encode()], exit_code)

    def close(self) -> None:
        """Close the reused sandbox, if one is alive."""
        if self._sandbox is not None:
            self._sandbox.__exit__(None, None, None)
            self._sandbox = None
            self._staged_root = None
            self._staged_fingerprint = None

run async 

run(script: SkillScript, args: dict[str, Any] | None = None, ctx: Any | None = None) -> Any

Run a skill script inside a LocalSandbox virtual filesystem.

Parameters:

Name Type Description Default
script SkillScript

The script to run; script.uri must point at a local file.

 required
args dict[str, Any] | None

Named arguments, marshalled with the same rules as LocalSkillScriptExecutor.

 None
ctx Any | None

Unused; accepted for protocol compatibility.

 None

Returns:

Type Description
Any

Combined stdout and stderr, formatted like local execution.

Raises:

Type Description
ValueError

If the script has no URI, or its type is unsupported here.
Source code in pydantic_ai_skills/sandboxes/localsandbox.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
async def run(
    self,
    script: SkillScript,
    args: dict[str, Any] | None = None,
    ctx: Any | None = None,
) -> Any:
    """Run a skill script inside a LocalSandbox virtual filesystem.

    Args:
        script: The script to run; ``script.uri`` must point at a local file.
        args: Named arguments, marshalled with the same rules as
            [`LocalSkillScriptExecutor`][pydantic_ai_skills.LocalSkillScriptExecutor].
        ctx: Unused; accepted for protocol compatibility.

    Returns:
        Combined stdout and stderr, formatted like local execution.

    Raises:
        ValueError: If the script has no URI, or its type is unsupported here.
    """
    del ctx  # Required by the SkillScriptExecutor protocol; unused by this backend.

    if script.uri is None:
        raise ValueError(f"Script '{script.name}' has no URI for sandbox execution")

    script_path = Path(script.uri).resolve()
    skill_root = skill_root_for(script)
    suffix = script_path.suffix.lower()
    remote_path = f'{self.workdir}/{script_path.relative_to(skill_root).as_posix()}'
    # cwd is the script's own directory, matching LocalSkillScriptExecutor.
    cwd = str(PurePosixPath(remote_path).parent)

    # Validated before provisioning, so an unsupported script never starts a sandbox.
    if suffix != '.py' and suffix not in _SHELL_INTERPRETERS:
        raise ValueError(
            f"Script '{script.name}' has unsupported type '{suffix}' for LocalSandbox. "
            f'Supported: .py (Pyodide), {", ".join(sorted(_SHELL_INTERPRETERS))} (just-bash).'
        )

    if not self._reuse_sandbox:
        return await self._execute(skill_root, script_path, remote_path, cwd, suffix, args)

    # One sandbox serving concurrent runs has to serialize them: a second run
    # switching skills would otherwise close the sandbox the first is still
    # using, and both would share a filesystem mid-execution anyway.
    async with self._reuse_lock:
        return await self._execute(skill_root, script_path, remote_path, cwd, suffix, args)

close 

close() -> None

Close the reused sandbox, if one is alive.

Source code in pydantic_ai_skills/sandboxes/localsandbox.py 
351
352
353
354
355
356
357
def close(self) -> None:
    """Close the reused sandbox, if one is alive."""
    if self._sandbox is not None:
        self._sandbox.__exit__(None, None, None)
        self._sandbox = None
        self._staged_root = None
        self._staged_fingerprint = None