Python: Add Hyperlight CodeAct package and docs (#5185)

* initial work on code_mode * updated samples * updates to codeact * udpated codeact * Draft CodeAct ADR and sample updates Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * initial implementation and adr and feature * Python: Limit Hyperlight wasm backend to Python <3.14 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Fix CI for Hyperlight CodeAct PR Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Run Hyperlight integration when available Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Address Hyperlight review feedback Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Simplify Hyperlight file mount inputs Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Accept Path host paths in Hyperlight mounts Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Python: Fix Hyperlight mount typing for CI Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * temp run integration test * Python: Strengthen Hyperlight real sandbox tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * added additional tests * Python: Simplify Hyperlight CodeAct API Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * set tests as non-integration * Retry Hyperlight allowed-domain registration Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Gate Hyperlight integration tests by runtime support Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix Hyperlight skip test on Python 3.14 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Delay Hyperlight runtime probe until test execution Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Relax Hyperlight Windows integration stdout assertion Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Scan Hyperlight output directory for artifacts Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Retry Hyperlight output artifact collection Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Harden Hyperlight integration output assertions Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Retry Hyperlight read-back check in integration test Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Simplify Hyperlight integration write assertion Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Avoid pathlib in Hyperlight integration sandbox Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Use socket network check in Hyperlight sandbox Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Replace blocked Azure AI Search blog link Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Clarify Hyperlight guest stdlib limits Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Use _socket in Hyperlight integration sandbox Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Handle Hyperlight mounted file paths Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Broaden Hyperlight sandbox path fallbacks Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Search Hyperlight guest mounts recursively Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Split Hyperlight mount coverage Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Split Hyperlight live network tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix Hyperlight file-write test on Windows Enable the sandbox filesystem by providing a workspace_root so /output is mounted. Remove os.path.exists assertion (unsupported in WASM guest) and fix Content data assertion to use .uri. Skip the network integration test on Windows where the WASM sandbox lacks the encodings.idna codec. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Address PR review: ADR intro, manual wiring sample, doc clarifications - Add CodeAct introduction section to ADR for unfamiliar readers - Clarify 'less runtime efficient' con with specific overhead description - Add note in Python impl doc clarifying ADR vs impl doc split - Explain why before_run hooks must be per-run (CRUD, concurrency, approval) - Rename code_interpreter variable to codeact in E2E sample - Add manual static wiring sample (codeact_manual_wiring.py) - Add 'when to use which pattern' guidance to samples README Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Address PR #5185 review comments and add .NET CodeAct design doc - Fix async callback: _make_sandbox_callback returns sync wrapper with thread + asyncio.run() bridge (was broken with real Wasm FFI) - Fix stale output: clear output_dir before each sandbox.run() call - Fix blocking event loop: _run_code now async with asyncio.to_thread() - Revert _agents.py options['tools'] injection (unnecessary; provider uses context.extend_tools()) - Revert SessionContext.options docstring back to read-only - Add real-sandbox test fixtures (shared/restored/fresh) - Add 8 new real-sandbox tests for callback round-trip, stale output, event loop non-blocking, basic execution, stdout/stderr, errors, snapshot/restore, and tool registration - Add comprehensive .NET HyperlightCodeActProvider design document Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Update hyperlight README with code snippets and remove Public API section Replace bare export list with Quick Start code examples covering the context provider, standalone tool, manual static wiring, and file mounts / network access patterns. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-06-16 21:04:09 +08:00 · 2026-04-17 02:49:44 +02:00
parent dbf935b4e3
commit b03cb324d5
25 changed files with 4176 additions and 9 deletions
@@ -0,0 +1,24 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+import importlib.metadata
+
+from ._execute_code_tool import HyperlightExecuteCodeTool
+from ._provider import HyperlightCodeActProvider
+from ._types import AllowedDomain, AllowedDomainInput, FileMount, FileMountInput
+
+try:
+    __version__ = importlib.metadata.version(__name__)
+except importlib.metadata.PackageNotFoundError:
+    __version__ = "0.0.0"
+
+__all__ = [
+    "AllowedDomain",
+    "AllowedDomainInput",
+    "FileMount",
+    "FileMountInput",
+    "HyperlightCodeActProvider",
+    "HyperlightExecuteCodeTool",
+    "__version__",
+]
@@ -0,0 +1,860 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+import ast
+import asyncio
+import copy
+import mimetypes
+import shutil
+import threading
+import time
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from tempfile import TemporaryDirectory
+from typing import Annotated, Any, Protocol, TypeGuard, cast
+from urllib.parse import urlparse
+
+from agent_framework import Content, FunctionTool
+from agent_framework._tools import ApprovalMode, normalize_tools
+from pydantic import BaseModel, Field
+
+from ._instructions import build_codeact_instructions, build_execute_code_description
+from ._types import AllowedDomain, AllowedDomainInput, FileMount, FileMountHostPath, FileMountInput
+
+DEFAULT_HYPERLIGHT_BACKEND = "wasm"
+DEFAULT_HYPERLIGHT_MODULE = "python_guest.path"
+EXECUTE_CODE_INPUT_DESCRIPTION = "Python code to execute in an isolated Hyperlight sandbox."
+OUTPUT_FILE_RETRY_ATTEMPTS = 10
+OUTPUT_FILE_RETRY_DELAY_SECONDS = 0.1
+
+
+class _ExecuteCodeInput(BaseModel):
+    code: Annotated[str, Field(description=EXECUTE_CODE_INPUT_DESCRIPTION)]
+
+
+@dataclass(frozen=True, slots=True)
+class _StoredFileMount:
+    host_path: Path
+    mount_path: str
+
+
+@dataclass(frozen=True, slots=True)
+class _NormalizedFileMount:
+    host_path: Path
+    mount_path: str
+    path_signature: tuple[tuple[str, int, int], ...]
+
+
+@dataclass(frozen=True, slots=True)
+class _RunConfig:
+    backend: str
+    module: str | None
+    module_path: str | None
+    approval_mode: ApprovalMode
+    tools: tuple[FunctionTool, ...]
+    workspace_root: Path | None
+    workspace_signature: tuple[tuple[str, int, int], ...]
+    file_mounts: tuple[_NormalizedFileMount, ...]
+    allowed_domains: tuple[AllowedDomain, ...]
+
+    @property
+    def mounted_paths(self) -> tuple[str, ...]:
+        return tuple(_display_mount_path(mount.mount_path) for mount in self.file_mounts)
+
+    @property
+    def filesystem_enabled(self) -> bool:
+        return self.workspace_root is not None or bool(self.file_mounts)
+
+    def cache_key(self) -> tuple[Any, ...]:
+        return (
+            self.backend,
+            self.module,
+            self.module_path,
+            self.approval_mode,
+            tuple((tool_obj.name, id(tool_obj)) for tool_obj in self.tools),
+            str(self.workspace_root) if self.workspace_root is not None else None,
+            self.workspace_signature,
+            tuple((mount.mount_path, str(mount.host_path), mount.path_signature) for mount in self.file_mounts),
+            tuple((allowed_domain.target, allowed_domain.methods) for allowed_domain in self.allowed_domains),
+        )
+
+
+class SandboxRuntime(Protocol):
+    def execute(self, *, config: _RunConfig, code: str) -> list[Content]: ...
+
+
+@dataclass
+class _SandboxEntry:
+    sandbox: Any
+    snapshot: Any
+    input_dir: TemporaryDirectory[str] | None
+    output_dir: TemporaryDirectory[str] | None
+    lock: threading.RLock
+
+
+def _load_sandbox_class() -> type[Any]:
+    try:
+        from hyperlight_sandbox import Sandbox
+    except ModuleNotFoundError as exc:
+        raise ModuleNotFoundError(
+            "Hyperlight support requires `hyperlight-sandbox`, `hyperlight-sandbox-python-guest`, "
+            "and a compatible backend package such as `hyperlight-sandbox-backend-wasm`."
+        ) from exc
+
+    return Sandbox
+
+
+def _passthrough_result_parser(result: Any) -> str:
+    return repr(result)
+
+
+def _collect_tools(*tool_groups: Any) -> list[FunctionTool]:
+    tools_by_name: dict[str, FunctionTool] = {}
+
+    for tool_group in tool_groups:
+        normalized_group = normalize_tools(tool_group)
+        for tool_obj in normalized_group:
+            if not isinstance(tool_obj, FunctionTool):
+                continue
+            if tool_obj.name == "execute_code":
+                continue
+            tools_by_name.pop(tool_obj.name, None)
+            tools_by_name[tool_obj.name] = tool_obj
+
+    return list(tools_by_name.values())
+
+
+def _resolve_execute_code_approval_mode(
+    *,
+    base_approval_mode: ApprovalMode,
+    tools: Sequence[FunctionTool],
+) -> ApprovalMode:
+    if base_approval_mode == "always_require":
+        return "always_require"
+
+    if any(tool_obj.approval_mode == "always_require" for tool_obj in tools):
+        return "always_require"
+
+    return "never_require"
+
+
+def _resolve_existing_path(value: str | Path) -> Path:
+    return Path(value).expanduser().resolve(strict=True)
+
+
+def _resolve_workspace_root(value: str | Path | None) -> Path | None:
+    if value is None:
+        return None
+
+    resolved_path = _resolve_existing_path(value)
+    if not resolved_path.is_dir():
+        raise ValueError("workspace_root must point to an existing directory.")
+    return resolved_path
+
+
+def _is_file_mount_pair(value: Any) -> TypeGuard[FileMount | tuple[FileMountHostPath, str]]:
+    if not isinstance(value, tuple):
+        return False
+
+    value_tuple = cast(tuple[object, ...], value)
+    if len(value_tuple) != 2:
+        return False
+
+    host_path, mount_path = value_tuple
+    return isinstance(host_path, (str, Path)) and isinstance(mount_path, str)
+
+
+def _normalize_file_mount_input(file_mount: FileMountInput) -> _StoredFileMount:
+    host_path: FileMountHostPath
+    mount_path: str
+    if isinstance(file_mount, str):
+        host_path = file_mount
+        mount_path = file_mount
+    else:
+        host_path = file_mount[0]
+        mount_path = file_mount[1]
+
+    return _StoredFileMount(
+        host_path=_resolve_existing_path(host_path),
+        mount_path=_normalize_mount_path(mount_path),
+    )
+
+
+def _normalize_domain(target: str) -> str:
+    candidate = target.strip()
+    if not candidate:
+        raise ValueError("Allowed domain entries must not be empty.")
+
+    parsed = urlparse(candidate if "://" in candidate else f"//{candidate}")
+    normalized = (parsed.netloc or parsed.path).strip().rstrip("/")
+    if not normalized:
+        raise ValueError(f"Could not normalize allowed domain entry: {target!r}.")
+    return normalized.lower()
+
+
+def _normalize_http_method(method: str) -> str:
+    normalized = method.strip().upper()
+    if not normalized:
+        raise ValueError("HTTP method entries must not be empty.")
+    return normalized
+
+
+def _normalize_http_methods(methods: str | Sequence[str] | None) -> tuple[str, ...] | None:
+    if methods is None:
+        return None
+
+    normalized_methods = (
+        {_normalize_http_method(methods)}
+        if isinstance(methods, str)
+        else {_normalize_http_method(method) for method in methods}
+    )
+    if not normalized_methods:
+        raise ValueError("Allowed domain methods must not be empty when provided.")
+    return tuple(sorted(normalized_methods))
+
+
+def _is_allowed_domain_pair(value: Any) -> TypeGuard[tuple[str, str | Sequence[str]]]:
+    if not isinstance(value, tuple) or isinstance(value, AllowedDomain):
+        return False
+
+    value_tuple = cast(tuple[object, ...], value)
+    if len(value_tuple) != 2:
+        return False
+
+    target, methods = value_tuple
+    if not isinstance(target, str):
+        return False
+    if isinstance(methods, str):
+        return True
+    return isinstance(methods, Sequence)
+
+
+def _normalize_allowed_domain_input(allowed_domain: AllowedDomainInput) -> AllowedDomain:
+    if isinstance(allowed_domain, str):
+        return AllowedDomain(target=_normalize_domain(allowed_domain), methods=None)
+
+    if isinstance(allowed_domain, AllowedDomain):
+        return AllowedDomain(
+            target=_normalize_domain(allowed_domain.target),
+            methods=_normalize_http_methods(allowed_domain.methods),
+        )
+
+    target, methods = allowed_domain
+    return AllowedDomain(
+        target=_normalize_domain(target),
+        methods=_normalize_http_methods(methods),
+    )
+
+
+def _allowed_domain_registration_targets(*, target: str, expand_missing_scheme: bool) -> tuple[str, ...]:
+    if not expand_missing_scheme or "://" in target:
+        return (target,)
+    return (f"http://{target}", f"https://{target}")
+
+
+def _should_retry_allowed_domain_registration(
+    *,
+    error: RuntimeError,
+    allowed_domains: Sequence[AllowedDomain],
+) -> bool:
+    message = str(error).lower()
+    return "invalid url for network permission" in message and any(
+        "://" not in domain.target for domain in allowed_domains
+    )
+
+
+def _normalize_mount_path(mount_path: str) -> str:
+    raw_path = mount_path.strip().replace("\\", "/")
+    if not raw_path:
+        raise ValueError("mount_path must not be empty.")
+
+    pure_path = PurePosixPath(raw_path)
+    parts = [part for part in pure_path.parts if part not in {"", "/", "."}]
+    if parts and parts[0] == "input":
+        parts = parts[1:]
+    if any(part == ".." for part in parts):
+        raise ValueError("mount_path must stay within /input.")
+    if not parts:
+        raise ValueError("mount_path must point to a concrete path under /input.")
+    return "/".join(parts)
+
+
+def _display_mount_path(mount_path: str) -> str:
+    return f"/input/{mount_path}"
+
+
+def _path_tree_signature(path: Path) -> tuple[tuple[str, int, int], ...]:
+    if path.is_file():
+        stat = path.stat()
+        return ((path.name, int(stat.st_size), int(stat.st_mtime_ns)),)
+
+    entries: list[tuple[str, int, int]] = []
+    for candidate in sorted(path.rglob("*"), key=lambda value: value.as_posix()):
+        try:
+            stat = candidate.stat()
+        except FileNotFoundError:
+            continue
+        relative_path = candidate.relative_to(path).as_posix()
+        size = int(stat.st_size) if candidate.is_file() else 0
+        entries.append((relative_path, size, int(stat.st_mtime_ns)))
+    return tuple(entries)
+
+
+def _copy_path(source: Path, destination: Path) -> None:
+    if source.is_dir():
+        destination.mkdir(parents=True, exist_ok=True)
+        for child in sorted(source.iterdir(), key=lambda value: value.name):
+            _copy_path(child, destination / child.name)
+        return
+
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(source, destination)
+
+
+def _populate_input_dir(*, config: _RunConfig, input_root: Path) -> None:
+    if config.workspace_root is not None:
+        for child in sorted(config.workspace_root.iterdir(), key=lambda value: value.name):
+            _copy_path(child, input_root / child.name)
+
+    for mount in config.file_mounts:
+        _copy_path(mount.host_path, input_root / mount.mount_path)
+
+
+def _create_file_content(file_path: Path, *, relative_path: str) -> Content:
+    media_type = mimetypes.guess_type(file_path.name)[0] or "application/octet-stream"
+    return Content.from_data(
+        data=file_path.read_bytes(),
+        media_type=media_type,
+        additional_properties={"path": f"/output/{relative_path}"},
+    )
+
+
+def _normalize_output_relative_path(*, output_file: object, root: Path) -> str | None:
+    candidate_path = Path(str(output_file))
+    if candidate_path.is_absolute():
+        try:
+            return candidate_path.relative_to(root).as_posix()
+        except ValueError:
+            return None
+
+    raw_path = str(output_file).replace("\\", "/")
+    pure_path = PurePosixPath(raw_path)
+    parts = [part for part in pure_path.parts if part not in {"", "/", "."}]
+    if parts and parts[0] == "output":
+        parts = parts[1:]
+    if not parts or any(part == ".." for part in parts):
+        return None
+    return "/".join(parts)
+
+
+def _collect_output_relative_paths(*, sandbox: Any, root: Path) -> set[str]:
+    relative_paths: set[str] = set()
+
+    if hasattr(sandbox, "get_output_files"):
+        try:
+            output_files = cast(Sequence[object], sandbox.get_output_files())
+        except Exception:
+            output_files = ()
+
+        for output_file in output_files:
+            if (relative_path := _normalize_output_relative_path(output_file=output_file, root=root)) is not None:
+                relative_paths.add(relative_path)
+
+    for host_path in root.rglob("*"):
+        if host_path.is_file():
+            relative_paths.add(host_path.relative_to(root).as_posix())
+
+    return relative_paths
+
+
+def _parse_output_files(
+    *,
+    sandbox: Any,
+    output_dir: TemporaryDirectory[str] | None,
+    expect_output_files: bool,
+) -> list[Content]:
+    if output_dir is None:
+        return []
+
+    root = Path(output_dir.name)
+
+    for attempt in range(OUTPUT_FILE_RETRY_ATTEMPTS):
+        relative_paths = _collect_output_relative_paths(sandbox=sandbox, root=root)
+        missing_files = expect_output_files and not relative_paths
+        contents: list[Content] = []
+
+        for relative_path in sorted(relative_paths):
+            host_path = root.joinpath(*PurePosixPath(relative_path).parts)
+            if not host_path.is_file():
+                missing_files = True
+                continue
+            try:
+                contents.append(_create_file_content(host_path, relative_path=relative_path))
+            except PermissionError:
+                missing_files = True
+
+        if not missing_files or attempt == OUTPUT_FILE_RETRY_ATTEMPTS - 1:
+            return contents
+
+        time.sleep(OUTPUT_FILE_RETRY_DELAY_SECONDS)
+
+    return []
+
+
+def _build_execution_contents(
+    *,
+    result: Any,
+    sandbox: Any,
+    output_dir: TemporaryDirectory[str] | None,
+    code: str,
+) -> list[Content]:
+    success = bool(getattr(result, "success", False))
+    stdout = str(getattr(result, "stdout", "") or "").replace("\r\n", "\n") or None
+    stderr = str(getattr(result, "stderr", "") or "").replace("\r\n", "\n") or None
+    outputs: list[Content] = []
+
+    if stdout is not None:
+        outputs.append(Content.from_text(stdout, raw_representation=result))
+
+    outputs.extend(
+        _parse_output_files(
+            sandbox=sandbox,
+            output_dir=output_dir,
+            expect_output_files="/output" in code,
+        )
+    )
+
+    if success:
+        if stderr is not None:
+            outputs.append(Content.from_text(stderr, raw_representation=result))
+        if not outputs:
+            outputs.append(Content.from_text("Code executed successfully without output."))
+        return [Content.from_code_interpreter_tool_result(outputs=outputs, raw_representation=result)]
+
+    error_details = stderr or "Unknown sandbox error"
+    outputs.append(
+        Content.from_error(
+            message="Execution error",
+            error_details=error_details,
+            raw_representation=result,
+        )
+    )
+    return [Content.from_code_interpreter_tool_result(outputs=outputs, raw_representation=result)]
+
+
+def _make_sandbox_callback(tool_obj: FunctionTool) -> Callable[..., Any]:
+    sandbox_tool = copy.copy(tool_obj)
+    sandbox_tool.result_parser = _passthrough_result_parser
+
+    def _callback(**kwargs: Any) -> Any:
+        async def _invoke() -> list[Content]:
+            return await sandbox_tool.invoke(arguments=kwargs)
+
+        # FunctionTool.invoke() is always async. The real Hyperlight backend invokes
+        # registered callbacks synchronously via FFI, so this must be a sync function.
+        # We run the async call on a dedicated thread to avoid conflicts with any
+        # event loop that may be running on the current thread.
+        result_box: list[Any] = [None]
+        error_box: list[BaseException] = []
+
+        def _run() -> None:
+            try:
+                result_box[0] = asyncio.run(_invoke())
+            except BaseException as exc:
+                error_box.append(exc)
+
+        worker = threading.Thread(target=_run)
+        worker.start()
+        worker.join()
+        if error_box:
+            raise error_box[0]
+        contents: list[Content] = result_box[0]
+
+        values: list[Any] = []
+        for content in contents:
+            if content.type == "text" and content.text is not None:
+                try:
+                    values.append(ast.literal_eval(content.text))
+                except (SyntaxError, ValueError):
+                    values.append(content.text)
+                continue
+
+            values.append(content.to_dict())
+
+        if len(values) == 1:
+            return values[0]
+        return values
+
+    return _callback
+
+
+def _clear_directory(output_dir: TemporaryDirectory[str] | None) -> None:
+    """Remove all contents of the output directory without deleting the directory itself."""
+    if output_dir is None:
+        return
+    root = Path(output_dir.name)
+    for child in root.iterdir():
+        try:
+            if child.is_symlink() or child.is_file():
+                child.unlink()
+            elif child.is_dir():
+                shutil.rmtree(child, ignore_errors=True)
+        except (FileNotFoundError, PermissionError):
+            pass
+
+
+class _SandboxRegistry:
+    def __init__(self) -> None:
+        self._entries: dict[tuple[Any, ...], _SandboxEntry] = {}
+        self._entries_lock = threading.RLock()
+
+    def execute(self, *, config: _RunConfig, code: str) -> list[Content]:
+        """Execute code in a cached sandbox matching the given config.
+
+        Entries are keyed by ``config.cache_key()``. Concurrent calls with the same
+        key are serialized by the entry lock so they never race, but they share the
+        same sandbox instance. For true parallel execution, use distinct provider
+        instances or configs that produce different cache keys.
+        """
+        cache_key = config.cache_key()
+        with self._entries_lock:
+            entry = self._entries.get(cache_key)
+            if entry is None:
+                entry = self._create_entry(config)
+                self._entries[cache_key] = entry
+
+        with entry.lock:
+            entry.sandbox.restore(entry.snapshot)
+            _clear_directory(entry.output_dir)
+            result = entry.sandbox.run(code=code)
+            return _build_execution_contents(
+                result=result,
+                sandbox=entry.sandbox,
+                output_dir=entry.output_dir,
+                code=code,
+            )
+
+    def _create_entry(self, config: _RunConfig) -> _SandboxEntry:
+        input_dir_handle = TemporaryDirectory() if config.filesystem_enabled else None
+        output_dir_handle = TemporaryDirectory() if config.filesystem_enabled else None
+
+        if input_dir_handle is not None:
+            _populate_input_dir(config=config, input_root=Path(input_dir_handle.name))
+
+        sandbox_cls = _load_sandbox_class()
+
+        def _create_sandbox() -> Any:
+            try:
+                return sandbox_cls(
+                    backend=config.backend,
+                    module=config.module,
+                    module_path=config.module_path,
+                    input_dir=input_dir_handle.name if input_dir_handle is not None else None,
+                    output_dir=output_dir_handle.name if output_dir_handle is not None else None,
+                )
+            except ImportError as exc:
+                raise RuntimeError(
+                    "The selected Hyperlight backend is not installed or not supported on this platform. "
+                    "Install a compatible backend package, such as `hyperlight-sandbox-backend-wasm`."
+                ) from exc
+
+        def _configure_sandbox(*, sandbox: Any, expand_missing_scheme: bool) -> None:
+            for tool_obj in config.tools:
+                sandbox.register_tool(tool_obj.name, _make_sandbox_callback(tool_obj))
+
+            for allowed_domain in config.allowed_domains:
+                for target in _allowed_domain_registration_targets(
+                    target=allowed_domain.target,
+                    expand_missing_scheme=expand_missing_scheme,
+                ):
+                    sandbox.allow_domain(
+                        target,
+                        methods=list(allowed_domain.methods) if allowed_domain.methods is not None else None,
+                    )
+
+        sandbox = _create_sandbox()
+        _configure_sandbox(sandbox=sandbox, expand_missing_scheme=False)
+
+        try:
+            sandbox.run("None")
+        except RuntimeError as exc:
+            if not _should_retry_allowed_domain_registration(error=exc, allowed_domains=config.allowed_domains):
+                raise
+
+            sandbox = _create_sandbox()
+            _configure_sandbox(sandbox=sandbox, expand_missing_scheme=True)
+            sandbox.run("None")
+
+        snapshot = sandbox.snapshot()
+        return _SandboxEntry(
+            sandbox=sandbox,
+            snapshot=snapshot,
+            input_dir=input_dir_handle,
+            output_dir=output_dir_handle,
+            lock=threading.RLock(),
+        )
+
+
+class HyperlightExecuteCodeTool(FunctionTool):
+    """Execute Python code inside a Hyperlight sandbox."""
+
+    def __init__(
+        self,
+        *,
+        tools: FunctionTool | Callable[..., Any] | Sequence[FunctionTool | Callable[..., Any]] | None = None,
+        approval_mode: ApprovalMode | None = None,
+        workspace_root: str | Path | None = None,
+        file_mounts: FileMountInput | Sequence[FileMountInput] | None = None,
+        allowed_domains: AllowedDomainInput | Sequence[AllowedDomainInput] | None = None,
+        backend: str = DEFAULT_HYPERLIGHT_BACKEND,
+        module: str | None = DEFAULT_HYPERLIGHT_MODULE,
+        module_path: str | None = None,
+        _registry: SandboxRuntime | None = None,
+    ) -> None:
+        super().__init__(
+            name="execute_code",
+            description=EXECUTE_CODE_INPUT_DESCRIPTION,
+            approval_mode="never_require",
+            func=self._run_code,
+            input_model=_ExecuteCodeInput,
+        )
+        self._state_lock = threading.RLock()
+        self._registry = _registry or _SandboxRegistry()
+        self._default_approval_mode: ApprovalMode = approval_mode or "never_require"
+        self._workspace_root = _resolve_workspace_root(workspace_root)
+        self._backend: str = backend
+        self._module: str | None = module
+        self._module_path: str | None = module_path
+        self._managed_tools: list[FunctionTool] = []
+        self._file_mounts: dict[str, _StoredFileMount] = {}
+        self._allowed_domains: dict[str, AllowedDomain] = {}
+
+        if tools is not None:
+            self.add_tools(tools)
+        if file_mounts is not None:
+            self.add_file_mounts(file_mounts)
+        if allowed_domains is not None:
+            self.add_allowed_domains(allowed_domains)
+
+        self._refresh_approval_mode()
+
+    @property
+    def description(self) -> str:
+        state_lock = getattr(self, "_state_lock", None)
+        if state_lock is None:
+            return str(self.__dict__.get("description", EXECUTE_CODE_INPUT_DESCRIPTION))
+
+        with state_lock:
+            allowed_domains = sorted(self._allowed_domains.values(), key=lambda value: value.target)
+            return build_execute_code_description(
+                tools=self._managed_tools,
+                filesystem_enabled=self._workspace_root is not None or bool(self._file_mounts),
+                workspace_enabled=self._workspace_root is not None,
+                mounted_paths=[_display_mount_path(mount.mount_path) for mount in self._file_mounts.values()],
+                allowed_domains=allowed_domains,
+            )
+
+    @description.setter
+    def description(self, value: str) -> None:
+        self.__dict__["description"] = value
+
+    def add_tools(
+        self,
+        tools: FunctionTool | Callable[..., Any] | Sequence[FunctionTool | Callable[..., Any]],
+    ) -> None:
+        """Add sandbox-managed tools to this execute_code surface."""
+        with self._state_lock:
+            combined_tools = _collect_tools(self._managed_tools, tools)
+            self._managed_tools = combined_tools
+            self._refresh_approval_mode()
+
+    def get_tools(self) -> list[FunctionTool]:
+        """Return the currently managed sandbox tools."""
+        with self._state_lock:
+            return list(self._managed_tools)
+
+    def remove_tool(self, name: str) -> None:
+        """Remove one managed sandbox tool by name."""
+        with self._state_lock:
+            remaining_tools = [tool_obj for tool_obj in self._managed_tools if tool_obj.name != name]
+            if len(remaining_tools) == len(self._managed_tools):
+                raise KeyError(f"No managed tool named {name!r} is registered.")
+            self._managed_tools = remaining_tools
+            self._refresh_approval_mode()
+
+    def clear_tools(self) -> None:
+        """Remove all managed sandbox tools."""
+        with self._state_lock:
+            self._managed_tools = []
+            self._refresh_approval_mode()
+
+    def add_file_mounts(self, file_mounts: FileMountInput | Sequence[FileMountInput]) -> None:
+        """Add one or more file mounts under `/input`.
+
+        A single string uses the same relative path on the host and in the sandbox.
+        Use a two-string tuple or `FileMount` when those paths differ.
+        """
+        if isinstance(file_mounts, str) or _is_file_mount_pair(file_mounts):
+            normalized_mounts = [_normalize_file_mount_input(file_mounts)]
+        else:
+            normalized_mounts = [
+                _normalize_file_mount_input(mount) for mount in cast(Sequence[FileMountInput], file_mounts)
+            ]
+
+        with self._state_lock:
+            for mount in normalized_mounts:
+                self._file_mounts[mount.mount_path] = mount
+
+    def get_file_mounts(self) -> list[FileMount]:
+        """Return the configured file mounts."""
+        with self._state_lock:
+            return [
+                FileMount(host_path=mount.host_path, mount_path=_display_mount_path(mount.mount_path))
+                for mount in self._file_mounts.values()
+            ]
+
+    def remove_file_mount(self, mount_path: str) -> None:
+        """Remove one file mount by its sandbox path."""
+        normalized_mount_path = _normalize_mount_path(mount_path)
+        with self._state_lock:
+            if normalized_mount_path not in self._file_mounts:
+                raise KeyError(f"No file mount exists for {mount_path!r}.")
+            del self._file_mounts[normalized_mount_path]
+
+    def clear_file_mounts(self) -> None:
+        """Remove all configured file mounts."""
+        with self._state_lock:
+            self._file_mounts.clear()
+
+    def add_allowed_domains(self, domains: AllowedDomainInput | Sequence[AllowedDomainInput]) -> None:
+        """Add one or more outbound allow-list entries."""
+        if isinstance(domains, (str, AllowedDomain)) or _is_allowed_domain_pair(domains):
+            normalized_domains = [_normalize_allowed_domain_input(domains)]
+        else:
+            normalized_domains = [
+                _normalize_allowed_domain_input(domain) for domain in cast(Sequence[AllowedDomainInput], domains)
+            ]
+
+        with self._state_lock:
+            for normalized_domain in normalized_domains:
+                self._allowed_domains[normalized_domain.target] = normalized_domain
+
+    def get_allowed_domains(self) -> list[AllowedDomain]:
+        """Return the configured outbound allow-list entries."""
+        with self._state_lock:
+            return sorted(self._allowed_domains.values(), key=lambda value: value.target)
+
+    def remove_allowed_domain(self, domain: str) -> None:
+        """Remove one outbound allow-list entry."""
+        normalized_domain = _normalize_domain(domain)
+        with self._state_lock:
+            if normalized_domain not in self._allowed_domains:
+                raise KeyError(f"No allowed domain exists for {domain!r}.")
+            del self._allowed_domains[normalized_domain]
+
+    def clear_allowed_domains(self) -> None:
+        """Remove all outbound allow-list entries."""
+        with self._state_lock:
+            self._allowed_domains.clear()
+
+    def build_instructions(self, *, tools_visible_to_model: bool) -> str:
+        """Build the current CodeAct instructions for this execute_code surface."""
+        config = self._build_run_config()
+        return build_codeact_instructions(
+            tools=config.tools,
+            tools_visible_to_model=tools_visible_to_model,
+        )
+
+    def create_run_tool(self) -> HyperlightExecuteCodeTool:
+        """Create a run-scoped snapshot of this execute_code surface."""
+        file_mounts = self.get_file_mounts()
+        allowed_domains = self.get_allowed_domains()
+
+        return HyperlightExecuteCodeTool(
+            tools=self.get_tools(),
+            approval_mode=self._default_approval_mode,
+            workspace_root=self._workspace_root,
+            file_mounts=file_mounts or None,
+            allowed_domains=allowed_domains or None,
+            backend=self._backend,
+            module=self._module,
+            module_path=self._module_path,
+            _registry=self._registry,
+        )
+
+    def build_serializable_state(self) -> dict[str, Any]:
+        """Return a JSON-serializable snapshot of the effective run state."""
+        config = self._build_run_config()
+        return {
+            "backend": config.backend,
+            "module": config.module,
+            "module_path": config.module_path,
+            "approval_mode": config.approval_mode,
+            "tool_names": [tool_obj.name for tool_obj in config.tools],
+            "filesystem_enabled": config.filesystem_enabled,
+            "workspace_root": str(config.workspace_root) if config.workspace_root is not None else None,
+            "file_mounts": [
+                {
+                    "host_path": str(mount.host_path),
+                    "mount_path": _display_mount_path(mount.mount_path),
+                }
+                for mount in config.file_mounts
+            ],
+            "network_enabled": bool(config.allowed_domains),
+            "allowed_domains": [
+                {
+                    "target": allowed_domain.target,
+                    "methods": list(allowed_domain.methods) if allowed_domain.methods is not None else None,
+                }
+                for allowed_domain in config.allowed_domains
+            ],
+        }
+
+    def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True) -> dict[str, Any]:
+        self.__dict__["description"] = self.description
+        return super().to_dict(exclude=exclude, exclude_none=exclude_none)
+
+    def _refresh_approval_mode(self) -> None:
+        self.approval_mode = _resolve_execute_code_approval_mode(
+            base_approval_mode=self._default_approval_mode,
+            tools=self._managed_tools,
+        )
+
+    def _build_run_config(self) -> _RunConfig:
+        with self._state_lock:
+            managed_tools = tuple(self._managed_tools)
+            workspace_root = self._workspace_root
+            stored_mounts = tuple(self._file_mounts.values())
+            allowed_domains = tuple(sorted(self._allowed_domains.values(), key=lambda value: value.target))
+            approval_mode = _resolve_execute_code_approval_mode(
+                base_approval_mode=self._default_approval_mode,
+                tools=managed_tools,
+            )
+
+        workspace_signature = _path_tree_signature(workspace_root) if workspace_root is not None else ()
+        normalized_mounts = tuple(
+            _NormalizedFileMount(
+                host_path=mount.host_path,
+                mount_path=mount.mount_path,
+                path_signature=_path_tree_signature(mount.host_path),
+            )
+            for mount in stored_mounts
+        )
+
+        return _RunConfig(
+            backend=self._backend,
+            module=self._module,
+            module_path=self._module_path,
+            approval_mode=approval_mode,
+            tools=managed_tools,
+            workspace_root=workspace_root,
+            workspace_signature=workspace_signature,
+            file_mounts=normalized_mounts,
+            allowed_domains=allowed_domains,
+        )
+
+    async def _run_code(self, *, code: str) -> list[Content]:
+        config = self._build_run_config()
+        return await asyncio.to_thread(self._registry.execute, config=config, code=code)
@@ -0,0 +1,126 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from agent_framework import FunctionTool
+
+from ._types import AllowedDomain
+
+
+def _format_tool_summaries(tools: Sequence[FunctionTool]) -> str:
+    if not tools:
+        return "- No tools are currently registered inside the sandbox."
+
+    lines: list[str] = []
+    for tool_obj in tools:
+        parameters = tool_obj.parameters().get("properties", {})
+        parameter_names = [name for name in parameters if isinstance(name, str)]
+        parameter_summary = ", ".join(parameter_names) if parameter_names else "none"
+        description = str(tool_obj.description or "").strip() or "No description provided."
+        lines.append(f"- `{tool_obj.name}`: {description} Parameters: {parameter_summary}.")
+    return "\n".join(lines)
+
+
+def _format_filesystem_capabilities(
+    *,
+    filesystem_enabled: bool,
+    workspace_enabled: bool,
+    mounted_paths: Sequence[str],
+) -> str:
+    if not filesystem_enabled:
+        return "Filesystem access is unavailable because no workspace root or file mounts are configured."
+
+    lines = ["Filesystem access is enabled."]
+    lines.append("Read files from `/input`.")
+    lines.append("Write generated artifacts to `/output`; returned files will be attached to the tool result.")
+
+    if workspace_enabled:
+        lines.append("The configured workspace root is available under `/input/`.")
+
+    if mounted_paths:
+        lines.append("Additional mounted paths:")
+        lines.extend(f"- `{mounted_path}`" for mounted_path in mounted_paths)
+    elif not workspace_enabled:
+        lines.append("No workspace root or explicit file mounts are currently configured.")
+
+    return "\n".join(lines)
+
+
+def _format_network_capabilities(
+    *,
+    allowed_domains: Sequence[AllowedDomain],
+) -> str:
+    if not allowed_domains:
+        return "Outbound network access is unavailable because no allow-listed targets are configured."
+
+    lines = ["Outbound network access is allowed only for these configured targets:"]
+    for allowed_domain in allowed_domains:
+        methods_text = (
+            ", ".join(allowed_domain.methods) if allowed_domain.methods else "all methods allowed by the backend"
+        )
+        lines.append(f"- `{allowed_domain.target}`: {methods_text}.")
+    return "\n".join(lines)
+
+
+def build_codeact_instructions(
+    *,
+    tools: Sequence[FunctionTool],
+    tools_visible_to_model: bool,
+) -> str:
+    """Build dynamic CodeAct instructions for the effective sandbox state."""
+    usage_note = (
+        "Some tools may also appear directly, but prefer `execute_code` whenever you need to combine Python "
+        "control flow with sandbox tool calls."
+        if tools_visible_to_model
+        else "Provider-owned sandbox tools are not exposed separately; use `execute_code` when you need them."
+    )
+
+    return f"""You have one primary tool: execute_code.
+
+Prefer one execute_code call per request when possible.
+Its tool description contains the current `call_tool(...)` guidance, sandbox
+tool registry, and capability limits.
+
+{usage_note}
+"""
+
+
+def build_execute_code_description(
+    *,
+    tools: Sequence[FunctionTool],
+    filesystem_enabled: bool,
+    workspace_enabled: bool,
+    mounted_paths: Sequence[str],
+    allowed_domains: Sequence[AllowedDomain],
+) -> str:
+    """Build the dynamic execute_code tool description for standalone usage."""
+    filesystem_text = _format_filesystem_capabilities(
+        filesystem_enabled=filesystem_enabled,
+        workspace_enabled=workspace_enabled,
+        mounted_paths=mounted_paths,
+    )
+    network_text = _format_network_capabilities(
+        allowed_domains=allowed_domains,
+    )
+
+    return f"""Execute Python in an isolated Hyperlight sandbox.
+
+Inside the sandbox, `call_tool(name, **kwargs)` is available as a built-in for
+registered host callbacks. Use the tool name as the first argument and keyword
+arguments only. Do not pass a dict or any other positional arguments after the
+tool name.
+
+Registered sandbox tools:
+{_format_tool_summaries(tools)}
+
+Filesystem capabilities:
+{filesystem_text}
+
+Network capabilities:
+{network_text}
+
+Prefer `execute_code` when you need to combine one or more `call_tool(...)`
+calls with Python control flow, loops, or post-processing.
+"""
@@ -0,0 +1,111 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+from pathlib import Path
+from typing import Any
+
+from agent_framework import AgentSession, ContextProvider, FunctionTool, SessionContext
+from agent_framework._tools import ApprovalMode
+
+from ._execute_code_tool import HyperlightExecuteCodeTool, SandboxRuntime
+from ._types import AllowedDomain, AllowedDomainInput, FileMount, FileMountInput
+
+
+class HyperlightCodeActProvider(ContextProvider):
+    """Inject a Hyperlight-backed CodeAct surface using provider-owned tools."""
+
+    DEFAULT_SOURCE_ID = "hyperlight_codeact"
+
+    def __init__(
+        self,
+        source_id: str = DEFAULT_SOURCE_ID,
+        *,
+        tools: FunctionTool | Callable[..., Any] | Sequence[FunctionTool | Callable[..., Any]] | None = None,
+        approval_mode: ApprovalMode | None = None,
+        workspace_root: str | Path | None = None,
+        file_mounts: FileMountInput | Sequence[FileMountInput] | None = None,
+        allowed_domains: AllowedDomainInput | Sequence[AllowedDomainInput] | None = None,
+        backend: str = "wasm",
+        module: str | None = "python_guest.path",
+        module_path: str | None = None,
+        _registry: SandboxRuntime | None = None,
+    ) -> None:
+        super().__init__(source_id)
+        self._execute_code_tool = HyperlightExecuteCodeTool(
+            tools=tools,
+            approval_mode=approval_mode,
+            workspace_root=workspace_root,
+            file_mounts=file_mounts,
+            allowed_domains=allowed_domains,
+            backend=backend,
+            module=module,
+            module_path=module_path,
+            _registry=_registry,
+        )
+
+    def add_tools(
+        self,
+        tools: FunctionTool | Callable[..., Any] | Sequence[FunctionTool | Callable[..., Any]],
+    ) -> None:
+        """Add provider-owned sandbox tools."""
+        self._execute_code_tool.add_tools(tools)
+
+    def get_tools(self) -> list[FunctionTool]:
+        """Return the provider-owned sandbox tools."""
+        return self._execute_code_tool.get_tools()
+
+    def remove_tool(self, name: str) -> None:
+        """Remove one provider-owned sandbox tool by name."""
+        self._execute_code_tool.remove_tool(name)
+
+    def clear_tools(self) -> None:
+        """Remove all provider-owned sandbox tools."""
+        self._execute_code_tool.clear_tools()
+
+    def add_file_mounts(self, file_mounts: FileMountInput | Sequence[FileMountInput]) -> None:
+        """Add provider-managed file mounts."""
+        self._execute_code_tool.add_file_mounts(file_mounts)
+
+    def get_file_mounts(self) -> list[FileMount]:
+        """Return the provider-managed file mounts."""
+        return self._execute_code_tool.get_file_mounts()
+
+    def remove_file_mount(self, mount_path: str) -> None:
+        """Remove one provider-managed file mount."""
+        self._execute_code_tool.remove_file_mount(mount_path)
+
+    def clear_file_mounts(self) -> None:
+        """Remove all provider-managed file mounts."""
+        self._execute_code_tool.clear_file_mounts()
+
+    def add_allowed_domains(self, domains: AllowedDomainInput | Sequence[AllowedDomainInput]) -> None:
+        """Add provider-managed outbound allow-list entries."""
+        self._execute_code_tool.add_allowed_domains(domains)
+
+    def get_allowed_domains(self) -> list[AllowedDomain]:
+        """Return the provider-managed outbound allow-list entries."""
+        return self._execute_code_tool.get_allowed_domains()
+
+    def remove_allowed_domain(self, domain: str) -> None:
+        """Remove one provider-managed outbound allow-list entry."""
+        self._execute_code_tool.remove_allowed_domain(domain)
+
+    def clear_allowed_domains(self) -> None:
+        """Remove all provider-managed outbound allow-list entries."""
+        self._execute_code_tool.clear_allowed_domains()
+
+    async def before_run(
+        self,
+        *,
+        agent: Any,
+        session: AgentSession | None,
+        context: SessionContext,
+        state: dict[str, Any],
+    ) -> None:
+        """Inject CodeAct instructions and a run-scoped execute_code tool before each run."""
+        run_tool = self._execute_code_tool.create_run_tool()
+        state[self.source_id] = run_tool.build_serializable_state()
+        context.extend_instructions(self.source_id, run_tool.build_instructions(tools_visible_to_model=False))
+        context.extend_tools(self.source_id, [run_tool])
@@ -0,0 +1,28 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from pathlib import Path
+from typing import NamedTuple, TypeAlias
+
+
+class FileMount(NamedTuple):
+    """Map a host file or directory into the sandbox input tree."""
+
+    host_path: str | Path
+    mount_path: str
+
+
+FileMountHostPath: TypeAlias = str | Path
+FileMountInput: TypeAlias = str | tuple[FileMountHostPath, str] | FileMount
+
+
+class AllowedDomain(NamedTuple):
+    """Allow outbound requests to one target, optionally restricted to specific HTTP methods."""
+
+    target: str
+    methods: tuple[str, ...] | None = None
+
+
+AllowedDomainInput: TypeAlias = str | tuple[str, str | Sequence[str]] | AllowedDomain
@@ -0,0 +1 @@
+