第12章 Omnigent 社区沙箱 Provider 指南

omnigent-acme/
├── pyproject.toml
└── src/
    └── omnigent/
        └── community/
            └── sandbox/
                └── acme/
                    ├── __init__.py
                    ├── config.py
                    ├── launcher.py
                    └── plugin.py

# pyproject.toml
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
 
[project]
name = "omnigent-acme"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["omnigent", "pydantic>=2", "click>=8"]
 
[project.entry-points."omnigent.sandbox_providers"]
acme = "omnigent.community.sandbox.acme.plugin:get_contribution"
 
[tool.setuptools.packages.find]
where = ["src"]
include = ["omnigent.community.sandbox.acme*"]
namespaces = true

# src/omnigent/community/sandbox/acme/config.py
from pydantic import BaseModel, ConfigDict, Field
 
 
class AcmeConfig(BaseModel):
    model_config = ConfigDict(extra="forbid")
 
    namespace: str = Field(default="sandboxes", min_length=1)

# src/omnigent/community/sandbox/acme/launcher.py
from omnigent.onboarding.sandboxes.base import (
    ExecModelHostLauncher,
    RemoteCommandResult,
)
from omnigent.onboarding.sandboxes.types import SandboxCapabilities
 
from .config import AcmeConfig
 
 
class AcmeSandboxLauncher(ExecModelHostLauncher):
    provider = "acme"
 
    def __init__(self, *, config: AcmeConfig) -> None:
        self.config = config
 
    @property
    def capabilities(self) -> SandboxCapabilities:
        return SandboxCapabilities(
            managed_launch=True,
            programmatic_terminate=True,
        )
 
    def prepare(self) -> None:
        """Load the SDK and verify provider credentials; safe to repeat."""
        raise NotImplementedError("Implement provider preflight")
 
    def provision(self, name: str) -> str:
        """Create a sandbox in self.config.namespace and return its id."""
        raise NotImplementedError("Implement sandbox creation")
 
    def run(
        self, sandbox_id: str, command: str, *, check: bool = True
    ) -> RemoteCommandResult:
        """Run a shell command; raise on nonzero exit only when check=True."""
        raise NotImplementedError("Implement shell execution")
 
    def terminate(self, sandbox_id: str) -> None:
        """Delete the sandbox; an already-absent sandbox is success."""
        raise NotImplementedError("Implement idempotent deletion")

• prepare() 校验工具和凭据，但不预配计算资源。必须可重复调用。可预期的运维故障应通过 click.ClickException 抛出，消息中带上 provider 名称和有用的、不含机密的信息。
• provision(name) -> str 返回一个不透明的沙箱 id，而不是 SandboxInfo 或 SDK 对象。如果在返回 id 之前创建失败，请清理已分配的部分资源；调用方无法清理一个从未收到的 id。
• run(..., check=True) -> RemoteCommandResult 返回 returncode、stdout 和 stderr。check=True 时，非零退出码要抛出 click.ClickException；check=False 时，返回失败结果。如果 SDK 接收 argv 列表，用 ["sh", "-c", command] 调用 shell，不要按空格拆分命令。不要记录原始命令字符串：host 启动命令中包含凭据。
• terminate(sandbox_id) 释放计算资源及其他 provider 侧资源。"已删除"应视为成功，但不要掩盖授权错误或临时故障。清理可以重试，且可能使用一个从未预配该沙箱的全新 launcher；要从 id 和已配置的凭据解析资源，而不是仅依赖内存中的句柄。不要要求清理实例先调用过 prepare()。

• 解析 $HOME 并创建 $HOME/workspace。
• 通过 materialize_workspace() 克隆所需的仓库。
• 在启动 host 前应用部署提供的 host_config。
• 启动 omnigent host --server ...，并在其环境中注入 host 身份和启动 token。
• 返回工作区的绝对路径：单仓库时是克隆目录，否则是父工作区。

from collections.abc import Callable, Sequence
 
from omnigent.onboarding.sandboxes.types import RepoWorkspace
 
 
def start_host(
    self,
    sandbox_id: str,
    *,
    token: str,
    host_id: str,
    host_name: str,
    server_url: str,
    repos: Sequence[RepoWorkspace] = (),
    host_config: dict[str, object] | None = None,
    on_stage: Callable[[str], None] | None = None,
) -> str:
    # Launch through your container entrypoint or control-plane API.
    raise NotImplementedError("Return the absolute in-sandbox workspace path")

# src/omnigent/community/sandbox/acme/plugin.py
from omnigent.onboarding.sandboxes.registry import (
    SandboxProviderContribution,
    SandboxProviderMetadata,
)
 
from .config import AcmeConfig
 
 
def get_contribution() -> SandboxProviderContribution:
    return SandboxProviderContribution(
        name="omnigent-acme",
        providers={
            "acme": SandboxProviderMetadata(
                name="acme",
                launcher_class=(
                    "omnigent.community.sandbox.acme.launcher:AcmeSandboxLauncher"
                ),
                config_model=AcmeConfig,
            ),
        },
    )

uv pip install -e .

sandbox:
  provider: acme
  server_url: https://your-server.example.com
  acme:
    namespace: sandboxes

omnigent sandbox create --provider acme \
  --server https://your-server.example.com \
  --repo-root /path/to/omnigent
 
omnigent sandbox connect --provider acme \
  --sandbox-id YOUR_SANDBOX_ID \
  --server https://your-server.example.com

from omnigent.onboarding.sandboxes.registry import instantiate, plugin_state
 
state = plugin_state()
print("Registered providers:", state.names())
print("Plugin load errors:", state.load_errors)
assert "acme" in state, state.load_errors
 
launcher = instantiate("acme", config={"namespace": "test-sandboxes"})
assert launcher.provider == "acme"
assert launcher.config.namespace == "test-sandboxes"
assert launcher.capabilities.managed_launch
assert not launcher.capabilities.cli_bootstrap