jianmu.execution

适用对象：工具执行扩展开发者 / 沙箱维护者 / 核心维护者 是否必读：按需 相关模块：jianmu.tool, jianmu.guard, jianmu.skill

1. 模块职责

jianmu.execution 负责“真正执行工具或技能”的运行时层，包括本地执行、Docker 沙箱执行、执行约束以及 runner 解析。

如果你只是在业务层声明工具，通常不需要先操作这里；如果你要控制工具运行环境、隔离策略或技能执行后端，这个模块就是主入口。

2. 适合查什么

• 执行约束：ExecutionConstraints
• 本地 / Docker 执行后端：LocalSandbox、DockerSandbox
• 工具执行门面：ToolRunner
• 执行事件上下文：ExecutionEventContext

3. 使用建议

• 普通工具作者通常先看 jianmu.tool，而不是直接从这里入手
• 需要控制执行环境、隔离策略或技能后端时，再进入这一层
• 希望按约束自动选择执行后端时，优先看 ToolRunner.from_execution_constraints()

4. 注意事项

• Docker 相关对象依赖宿主环境，不是“导入即能用”
• 应把这里看成执行基础设施层，而不是普通工具作者的第一入口

5. 最小示例

from jianmu.execution import ExecutionConstraints

constraints = ExecutionConstraints(
    mode="docker",
    docker_image="python:3.12-slim",
    timeout_s=30,
)

6. 常见入口

• 想声明沙箱约束：看 ExecutionConstraints
• 想直接用本地执行：看 LocalSandbox
• 想接 Docker 隔离：看 DockerSandbox
• 想按约束自动解析工具 runner：看 ToolRunner.from_execution_constraints()

7. API 参考

约束与协议

ExecutionConstraints dataclass

ExecutionConstraints(
    mode: Optional[str] = None,
    scope: Optional[str] = None,
    docker_image: Optional[str] = None,
    network: Optional[str] = None,
    memory: Optional[str] = None,
    cpus: Optional[float] = None,
    timeout_s: Optional[float] = None,
    tmpfs_noexec: bool = True,
)

Describe runtime limits and backend preferences for execution.

属性：

名称	类型	描述
mode	Optional[str]	Preferred execution backend such as local or docker.
scope	Optional[str]	Optional scope label indicating where the constraints apply.
docker_image	Optional[str]	Optional Docker image override.
network	Optional[str]	Optional Docker network mode.
memory	Optional[str]	Optional Docker memory limit.
cpus	Optional[float]	Optional Docker CPU limit.
timeout_s	Optional[float]	Optional timeout in seconds for the execution request.
tmpfs_noexec	bool	Whether Docker tmpfs mounts should disable execution.

JobResult dataclass

JobResult(
    stdout: str,
    stderr: str,
    exit_code: int,
    artifacts: Dict[str, bytes],
)

Represent the result of a sandboxed script execution.

属性：

名称	类型	描述
stdout	str	Captured standard output emitted by the script.
stderr	str	Captured standard error emitted by the script.
exit_code	int	Process exit status returned by the sandbox runtime.
artifacts	Dict[str, bytes]	Output files collected from the sandbox work directory.

SandboxRunnerProtocol

Bases: Protocol

Define the execution interface shared by local and Docker sandboxes.

run_tool async

run_tool(tool: Any, args: dict | None = None) -> Any

Execute a single tool call inside the sandbox backend.

参数：

名称	类型	描述	默认
tool	Any	Tool instance to execute.	必需
args	dict | None	Optional JSON-like arguments passed to the tool.	None

返回：

类型	描述
Any	Tool-specific return value.

源代码位于： jianmu/execution/base.py

async def run_tool(self, tool: Any, args: dict | None = None) -> Any:
    """Execute a single tool call inside the sandbox backend.

    Args:
        tool: Tool instance to execute.
        args: Optional JSON-like arguments passed to the tool.

    Returns:
        Tool-specific return value.
    """
    ...

run_job async

run_job(
    script: str,
    input_files: Optional[Dict[str, bytes]] = None,
    timeout_s: Optional[float] = None,
) -> JobResult

Run a standalone Python script with mounted inputs.

参数：

名称	类型	描述	默认
script	str	Python source code to execute.	必需
input_files	Optional[Dict[str, bytes]]	Optional input files keyed by relative path.	None
timeout_s	Optional[float]	Optional execution timeout in seconds.	None

返回：

类型	描述
JobResult	Captured process output plus any generated artifacts.

源代码位于： jianmu/execution/base.py

async def run_job(
    self,
    script: str,
    input_files: Optional[Dict[str, bytes]] = None,
    timeout_s: Optional[float] = None,
) -> JobResult:
    """Run a standalone Python script with mounted inputs.

    Args:
        script: Python source code to execute.
        input_files: Optional input files keyed by relative path.
        timeout_s: Optional execution timeout in seconds.

    Returns:
        Captured process output plus any generated artifacts.
    """
    ...

本地执行

LocalSandbox

Bases: SandboxRunnerProtocol

Local runner: execute tools directly in-process.

run_tool async

run_tool(tool: Any, args: dict | None = None) -> Any

Execute a tool directly in the current Python process.

参数：

名称	类型	描述	默认
tool	Any	Tool instance to invoke.	必需
args	dict | None	Optional tool arguments.	None

返回：

类型	描述
Any	Tool-specific return value.

源代码位于： jianmu/execution/local.py

async def run_tool(self, tool: Any, args: dict | None = None) -> Any:
    """Execute a tool directly in the current Python process.

    Args:
        tool: Tool instance to invoke.
        args: Optional tool arguments.

    Returns:
        Tool-specific return value.
    """
    return await tool.execute(args or {})

run_job async

run_job(
    script: str,
    input_files: Optional[Dict[str, bytes]] = None,
    timeout_s: Optional[float] = None,
) -> JobResult

Execute a standalone Python script on the local machine.

参数：

名称	类型	描述	默认
script	str	Python source code to run.	必需
input_files	Optional[Dict[str, bytes]]	Optional input files materialized in the temp workdir.	None
timeout_s	Optional[float]	Optional subprocess timeout in seconds.	None

返回：

类型	描述
JobResult	Captured process streams, exit code, and generated artifacts.

引发：

类型	描述
TimeoutExpired	If the operation fails.

源代码位于： jianmu/execution/local.py

async def run_job(
    self,
    script: str,
    input_files: Optional[Dict[str, bytes]] = None,
    timeout_s: Optional[float] = None,
) -> JobResult:
    """Execute a standalone Python script on the local machine.

    Args:
        script: Python source code to run.
        input_files: Optional input files materialized in the temp workdir.
        timeout_s: Optional subprocess timeout in seconds.

    Returns:
        Captured process streams, exit code, and generated artifacts.

    Raises:
        TimeoutExpired: If the operation fails.
    """
    tmp_dir = tempfile.mkdtemp(prefix="jianmu_local_job_")
    tmp_path = Path(tmp_dir)
    try:
        # Write script
        script_path = tmp_path / "main.py"
        with open(script_path, "w", encoding="utf-8") as f:
            f.write(script)

        # Write input files
        excluded_files = {"main.py"}
        if input_files:
            for name, content in input_files.items():
                safe_rel = _safe_relative_path(name)
                target = tmp_path / safe_rel
                target.parent.mkdir(parents=True, exist_ok=True)
                with open(target, "wb") as f:
                    f.write(content)
                excluded_files.add(safe_rel.as_posix())

        proc = await asyncio.create_subprocess_exec(
            os.sys.executable,
            str(script_path),
            cwd=tmp_dir,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        try:
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout_s)
        except asyncio.CancelledError:
            proc.kill()
            await proc.wait()
            raise
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            raise subprocess.TimeoutExpired([os.sys.executable, str(script_path)], timeout_s)

        artifacts = {}
        for entry in tmp_path.rglob("*"):
            if not entry.is_file():
                continue
            rel_name = entry.relative_to(tmp_path).as_posix()
            if rel_name in excluded_files:
                continue
            artifacts[rel_name] = entry.read_bytes()

        return JobResult(
            stdout=stdout.decode("utf-8", errors="replace"),
            stderr=stderr.decode("utf-8", errors="replace"),
            exit_code=proc.returncode,
            artifacts=artifacts,
        )
    finally:
        shutil.rmtree(tmp_dir, ignore_errors=True)

Docker 执行

DockerSandbox

DockerSandbox(
    constraints: ExecutionConstraints | None = None,
    *,
    skill_dir: str | Path | None = None,
    builtin_tool_names: list[str] | None = None,
)

Bases: SandboxRunnerProtocol

Sandbox runner backed by Docker.

属性：

名称	类型	描述
constraints		Execution constraints applied to Docker calls.
tool_runner		Docker-backed runner used for tool and job execution.

Create a sandbox facade over docker tool and skill runners.

参数：

名称	类型	描述	默认
constraints	ExecutionConstraints | None	Optional execution constraints for all docker calls.	None
skill_dir	str | Path | None	Optional skill directory used for tool discovery.	None
builtin_tool_names	list[str] | None	Optional builtin tool names exposed in docker mode.	None

源代码位于： jianmu/execution/docker.py

def __init__(
    self,
    constraints: ExecutionConstraints | None = None,
    *,
    skill_dir: str | Path | None = None,
    builtin_tool_names: list[str] | None = None,
):
    """Create a sandbox facade over docker tool and skill runners.

    Args:
        constraints: Optional execution constraints for all docker calls.
        skill_dir: Optional skill directory used for tool discovery.
        builtin_tool_names: Optional builtin tool names exposed in docker mode.
    """
    self.constraints = constraints or ExecutionConstraints(mode="docker")
    self.tool_runner = DockerToolRunner(
        skill_dir=Path(skill_dir) if skill_dir else None,
        image=self.constraints.docker_image,
        network=self.constraints.network,
        memory=self.constraints.memory,
        cpus=self.constraints.cpus,
        timeout_s=self.constraints.timeout_s,
        tmpfs_noexec=self.constraints.tmpfs_noexec,
        builtin_names=builtin_tool_names,
    )

run_tool async

run_tool(tool: Any, args: dict | None = None) -> Any

Execute a tool call through the docker tool runner.

参数：

名称	类型	描述	默认
tool	Any	The tool instance or proxy to run.	必需
args	dict | None	Optional dictionary of tool arguments. Defaults to None.	None

返回：

类型	描述
Any	The execution result of the tool.

源代码位于： jianmu/execution/docker.py

async def run_tool(self, tool: Any, args: dict | None = None) -> Any:
    """Execute a tool call through the docker tool runner.

    Args:
        tool: The tool instance or proxy to run.
        args: Optional dictionary of tool arguments. Defaults to None.

    Returns:
        The execution result of the tool.
    """
    return await self.tool_runner.run(tool, args or {})

run_job async

run_job(
    script: str,
    input_files: Dict[str, bytes] | None = None,
    timeout_s: float | None = None,
) -> JobResult

Execute a standalone Python script in Docker.

参数：

名称	类型	描述	默认
script	str	The Python script source code to run.	必需
input_files	Dict[str, bytes] | None	Optional dictionary mapping filenames to their byte content. Defaults to None.	None
timeout_s	float | None	Optional execution timeout in seconds. Defaults to None.	None

返回：

类型	描述
JobResult	A JobResult containing stdout, stderr, exit code, and generated artifacts.

源代码位于： jianmu/execution/docker.py

async def run_job(
    self,
    script: str,
    input_files: Dict[str, bytes] | None = None,
    timeout_s: float | None = None,
) -> JobResult:
    """Execute a standalone Python script in Docker.

    Args:
        script: The Python script source code to run.
        input_files: Optional dictionary mapping filenames to their byte content. Defaults to None.
        timeout_s: Optional execution timeout in seconds. Defaults to None.

    Returns:
        A JobResult containing stdout, stderr, exit code, and generated artifacts.
    """
    return await self.tool_runner.run_job(script, input_files, timeout_s)

close

close() -> None

Close cached persistent docker sessions owned by this sandbox.

源代码位于： jianmu/execution/docker.py

def close(self) -> None:
    """Close cached persistent docker sessions owned by this sandbox."""
    self.tool_runner.close()

工具执行门面

ExecutionEventContext dataclass

ExecutionEventContext(
    run_id: str | None = None,
    node_name: str | None = None,
    tool_call_id: str | None = None,
)

Optional correlation fields attached to execution.* events.

属性：

名称	类型	描述
run_id	str | None	Logical run/session identifier for the surrounding runtime.
node_name	str | None	Display name of the node that triggered execution.
tool_call_id	str | None	Provider or runtime tool-call correlation identifier.

ToolRunner

ToolRunner(
    mode: str,
    sandbox: Any,
    *,
    runtime_event_bus: RuntimeEventBus | None = None,
    timeout_s: float | None = None,
)

Unified facade for executing tools through local or docker backends.

属性：

名称	类型	描述
mode		Normalized backend mode such as local or docker.
_sandbox		Concrete sandbox implementation handling execution.
_runtime_event_bus		Optional runtime-semantic event bus for execution events.
_timeout_s		Optional wall-clock timeout for one tool execution.

Bind the runner to one concrete sandbox backend.

参数：

名称	类型	描述	默认
mode	str	Normalized backend mode such as local or docker.	必需
sandbox	Any	Concrete sandbox implementation used for execution.	必需
runtime_event_bus	RuntimeEventBus | None	Optional runtime-semantic event bus that receives execution.* lifecycle events for each tool invocation.	None
timeout_s	float | None	Optional wall-clock timeout for one tool execution.	None

源代码位于： jianmu/execution/runner.py

def __init__(
    self,
    mode: str,
    sandbox: Any,
    *,
    runtime_event_bus: RuntimeEventBus | None = None,
    timeout_s: float | None = None,
):
    """Bind the runner to one concrete sandbox backend.

    Args:
        mode: Normalized backend mode such as ``local`` or ``docker``.
        sandbox: Concrete sandbox implementation used for execution.
        runtime_event_bus: Optional runtime-semantic event bus that receives
            ``execution.*`` lifecycle events for each tool invocation.
        timeout_s: Optional wall-clock timeout for one tool execution.
    """
    self.mode = mode
    self._sandbox = sandbox
    self._runtime_event_bus = runtime_event_bus
    self._timeout_s = float(timeout_s) if timeout_s is not None else None

run async

run(
    tool: Any,
    args: Any,
    *,
    event_context: ExecutionEventContext | None = None,
) -> Any

Execute one tool call through the configured sandbox backend.

参数：

名称	类型	描述	默认
tool	Any	Tool instance or compatible callable wrapper being executed.	必需
args	Any	Normalized tool arguments passed to the sandbox backend.	必需
event_context	ExecutionEventContext | None	Optional correlation metadata copied onto emitted execution.* runtime events.	None

返回：

类型	描述
Any	The tool-specific execution result returned by the sandbox.

引发：

类型	描述
Exception	Re-raises backend execution failures after emitting execution.failed on the runtime event bus.

源代码位于： jianmu/execution/runner.py

async def run(
    self,
    tool: Any,
    args: Any,
    *,
    event_context: ExecutionEventContext | None = None,
) -> Any:
    """Execute one tool call through the configured sandbox backend.

    Args:
        tool: Tool instance or compatible callable wrapper being executed.
        args: Normalized tool arguments passed to the sandbox backend.
        event_context: Optional correlation metadata copied onto emitted
            ``execution.*`` runtime events.

    Returns:
        The tool-specific execution result returned by the sandbox.

    Raises:
        Exception: Re-raises backend execution failures after emitting
            ``execution.failed`` on the runtime event bus.
    """
    tool_name = getattr(tool, "name", type(tool).__name__)
    emit_runtime_event(
        self._runtime_event_bus,
        "execution.started",
        payload={
            "backend": self.mode,
            "tool": tool_name,
        },
        run_id=event_context.run_id if event_context else None,
        node_name=event_context.node_name if event_context else None,
        tool_call_id=event_context.tool_call_id if event_context else None,
    )
    try:
        if self._timeout_s is not None:
            result = await asyncio.wait_for(self._sandbox.run_tool(tool, args), timeout=self._timeout_s)
        else:
            result = await self._sandbox.run_tool(tool, args)
    except asyncio.TimeoutError as exc:
        emit_runtime_event(
            self._runtime_event_bus,
            "execution.failed",
            payload={
                "backend": self.mode,
                "tool": tool_name,
                "error": f"tool_timeout: timed out after {self._timeout_s}s",
                "timeout_s": self._timeout_s,
            },
            run_id=event_context.run_id if event_context else None,
            node_name=event_context.node_name if event_context else None,
            tool_call_id=event_context.tool_call_id if event_context else None,
        )
        raise TimeoutError(f"Tool '{tool_name}' timed out after {self._timeout_s}s") from exc
    except Exception as exc:
        emit_runtime_event(
            self._runtime_event_bus,
            "execution.failed",
            payload={
                "backend": self.mode,
                "tool": tool_name,
                "error": str(exc),
            },
            run_id=event_context.run_id if event_context else None,
            node_name=event_context.node_name if event_context else None,
            tool_call_id=event_context.tool_call_id if event_context else None,
        )
        raise
    emit_runtime_event(
        self._runtime_event_bus,
        "execution.completed",
        payload={
            "backend": self.mode,
            "tool": tool_name,
        },
        run_id=event_context.run_id if event_context else None,
        node_name=event_context.node_name if event_context else None,
        tool_call_id=event_context.tool_call_id if event_context else None,
    )
    return result

from_execution_constraints classmethod

from_execution_constraints(
    execution: Optional[ExecutionConstraints | dict] = None,
    *,
    builtin_names: Optional[Iterable[str]] = None,
    skill_dir: Optional[str] = None,
    runtime_event_bus: RuntimeEventBus | None = None,
) -> Optional["ToolRunner"]

Resolve a concrete tool runner from execution constraints and config.

参数：

名称	类型	描述	默认
execution	Optional[ExecutionConstraints | dict]	Optional execution constraints or raw dictionary payload.	None
builtin_names	Optional[Iterable[str]]	Optional builtin tool names used by Docker backends to expose local tool shims inside the container runtime.	None
skill_dir	Optional[str]	Optional skill directory used by Docker backends when resolving skill-relative execution context.	None
runtime_event_bus	RuntimeEventBus | None	Optional runtime-semantic event bus that receives execution.* lifecycle events from the resolved runner.	None

返回：

类型	描述
Optional['ToolRunner']	A configured ToolRunner instance, or None when no supported
Optional['ToolRunner']	execution mode can be resolved.

源代码位于： jianmu/execution/runner.py

@classmethod
def from_execution_constraints(
    cls,
    execution: Optional[ExecutionConstraints | dict] = None,
    *,
    builtin_names: Optional[Iterable[str]] = None,
    skill_dir: Optional[str] = None,
    runtime_event_bus: RuntimeEventBus | None = None,
) -> Optional["ToolRunner"]:
    """Resolve a concrete tool runner from execution constraints and config.

    Args:
        execution: Optional execution constraints or raw dictionary payload.
        builtin_names: Optional builtin tool names used by Docker backends to
            expose local tool shims inside the container runtime.
        skill_dir: Optional skill directory used by Docker backends when
            resolving skill-relative execution context.
        runtime_event_bus: Optional runtime-semantic event bus that receives
            ``execution.*`` lifecycle events from the resolved runner.

    Returns:
        A configured ``ToolRunner`` instance, or ``None`` when no supported
        execution mode can be resolved.
    """
    if isinstance(execution, dict):
        try:
            execution = ExecutionConstraints(**execution)
        except Exception:
            execution = None

    from jianmu.config.loader import get_config

    configured_mode = get_config().execution.tool_runner
    default_timeout_s = get_config().execution.timeout_s
    mode = _normalize_mode(execution.mode if execution else configured_mode)

    if mode == "docker":
        from jianmu.execution.docker import DockerSandbox

        sandbox = DockerSandbox(
            constraints=execution,
            skill_dir=skill_dir,
            builtin_tool_names=list(builtin_names or []),
        )
        return cls(
            mode="docker",
            sandbox=sandbox,
            runtime_event_bus=runtime_event_bus,
            timeout_s=execution.timeout_s if execution is not None else default_timeout_s,
        )

    if mode == "local":
        from jianmu.execution.local import LocalSandbox

        return cls(
            mode="local",
            sandbox=LocalSandbox(),
            runtime_event_bus=runtime_event_bus,
            timeout_s=execution.timeout_s if execution is not None else default_timeout_s,
        )

    return None