#!/usr/bin/env python3
"""
Shell command execution implementation for Klea MCP tools (ADR-0038).
Timeout and cancellation signal the command's process group (SIGTERM, then
SIGKILL). A descendant that detaches from the group (``setsid``/``setpgid``,
as some build tools do) cannot be reached by a portable group signal and may
survive.
See ``devdocs/system/mcp-permissions.md``.
File: klea_utils/mcp/tool_impls/run_command.py
Copyright 2026 Ankur Sinha
Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>
"""
import asyncio
import logging
import math
import os
import signal
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
#: Default per-command timeout in seconds.
DEFAULT_TIMEOUT_SECONDS = 30.0
#: Default ceiling for the per-command timeout, overridable via
#: :data:`MAX_TIMEOUT_ENV_VAR` (mirrors opencode's 10-minute bash cap).
DEFAULT_MAX_TIMEOUT_SECONDS = 600.0
#: Process environment variable that raises/lowers the timeout ceiling
#: (seconds). A process env var, not an env-file/config field: it is an
#: operational bound (like ``KLEA_LOG_LEVEL``), not app configuration.
MAX_TIMEOUT_ENV_VAR = "KLEA_RUN_COMMAND_MAX_TIMEOUT"
#: Default cap on captured characters per stream (stdout, stderr).
DEFAULT_MAX_OUTPUT_CHARS = 100_000
#: Grace period between SIGTERM and SIGKILL when killing a timed-out process
#: group.
KILL_GRACE_SECONDS = 3.0
#: Poll interval while waiting for a signalled process group to drain.
_GROUP_POLL_SECONDS = 0.05
def _shell_display(
command: str,
returncode: int | None,
stdout: str,
stderr: str,
error: str,
) -> dict[str, Any] | None:
"""Build the chat-facing ``display`` block for a command result.
Returns a self-describing display dict (ADR-0040 convention:
``{"mime", "data", "meta"}``) with MIME ``text/x-shell`` so the chat
renders one shell block. The block shows the command, stdout, stderr
(when non-empty) and the exit status, so the user can keep an eye on
what was run (``run_command`` is destructive). The full captured output
is included (already bounded per stream by ``max_output_chars``); the UI
collapses long blocks.
The block is omitted (returns ``None``) for a silent success: no stdout,
no stderr, and a zero exit. A non-zero exit, captured output, or a
call-level ``error`` always produces a block.
:param command: The command string that was run.
:param returncode: Exit status, or ``None`` when the command never ran.
:param stdout: Captured standard output.
:param stderr: Captured standard error.
:param error: Call-level error message (timeout/denied/spawn), or ``""``.
:returns: The display dict, or ``None`` when there is nothing to show.
"""
if not stdout and not stderr and not error and returncode == 0:
return None
lines = [f"$ {command}"]
if stdout:
lines.append(stdout.rstrip("\n"))
if stderr:
lines.append("--- stderr ---")
lines.append(stderr.rstrip("\n"))
if error:
lines.append(f"--- error ---\n{error}")
if returncode is not None:
lines.append(f"[exit {returncode}]")
return {
"mime": "text/x-shell",
"data": "\n".join(lines),
"meta": {
"command": command,
"returncode": returncode,
},
}
def _result(
command: str,
working_directory: str | None,
*,
returncode: int | None = None,
stdout: str = "",
stderr: str = "",
truncated: bool = False,
error: str = "",
) -> dict[str, Any]:
"""Build the standard command result dict.
The dict also carries a chat-facing ``display`` field (ADR-0040
self-describing convention, ``{"mime", "data", "meta"}`` or ``None``)
so the tools caller can render the command, its output and its exit
status in the chat without a Klea-specific schema.
:param command: The command string that was run.
:param working_directory: Working directory used, or ``None`` when the
command never ran.
:param returncode: Exit status, or ``None`` when the command never ran.
:param stdout: Captured standard output (possibly truncated).
:param stderr: Captured standard error (possibly truncated).
:param truncated: Whether either stream was truncated.
:param error: Empty when the command produced a result (any exit code);
a message only for call-level failures (timeout, denied working
directory, spawn failure) or a rejected argument.
:returns: The result dict returned to the MCP wrapper.
"""
return {
"command": command,
"working_directory": working_directory,
"returncode": returncode,
"stdout": stdout,
"stderr": stderr,
"truncated": truncated,
"error": error,
"display": _shell_display(command, returncode, stdout, stderr, error),
}
[docs]
def max_timeout_seconds() -> float:
"""Return the effective timeout ceiling in seconds.
Reads :data:`MAX_TIMEOUT_ENV_VAR` from the process environment (always a
string) and casts it to a float; an unset, non-numeric, non-finite, or
non-positive value falls back to :data:`DEFAULT_MAX_TIMEOUT_SECONDS` with
a warning.
:returns: The ceiling for ``timeout_seconds``.
"""
raw = os.environ.get(MAX_TIMEOUT_ENV_VAR)
if raw is None:
return DEFAULT_MAX_TIMEOUT_SECONDS
try:
value = float(raw)
except ValueError:
logger.warning(
f"Invalid {MAX_TIMEOUT_ENV_VAR}={raw!r}; "
f"using {DEFAULT_MAX_TIMEOUT_SECONDS} s"
)
return DEFAULT_MAX_TIMEOUT_SECONDS
if not math.isfinite(value) or value <= 0:
logger.warning(
f"Invalid (non-finite or non-positive) {MAX_TIMEOUT_ENV_VAR}={raw!r}; "
f"using {DEFAULT_MAX_TIMEOUT_SECONDS} s"
)
return DEFAULT_MAX_TIMEOUT_SECONDS
logger.debug(f"{MAX_TIMEOUT_ENV_VAR} = {value}")
return value
def _process_group_exists(pgid: int) -> bool:
"""Return whether any process remains in process group *pgid*.
Probes the group with the null signal: ``os.killpg`` raises
:class:`ProcessLookupError` when the group is empty.
:param pgid: Process group id to probe.
:returns: True when at least one process is still in the group.
"""
try:
os.killpg(pgid, 0)
except ProcessLookupError:
return False
except PermissionError: # pragma: no cover - the group is ours
return True
return True
async def _drain_process_group(pgid: int, timeout: float) -> bool:
"""Wait up to *timeout* seconds for process group *pgid* to become empty.
:param pgid: Process group id to watch.
:param timeout: Maximum seconds to wait.
:returns: True when the group drained within *timeout*.
"""
loop = asyncio.get_running_loop()
deadline = loop.time() + timeout
while _process_group_exists(pgid):
if loop.time() >= deadline:
return False
await asyncio.sleep(_GROUP_POLL_SECONDS)
return True
async def _terminate_process_group(process: asyncio.subprocess.Process) -> None:
"""Terminate a process and its children (its process group).
Sends SIGTERM to the whole group, waits :data:`KILL_GRACE_SECONDS` for the
*group* to drain, then SIGKILLs any survivors. Waiting on the group rather
than only the direct child matters: a descendant can outlive the shell that
spawned it when it ignores or defers SIGTERM, and the group is gone only
once its last member exits. Best-effort: an already-exited process or
group is ignored.
A descendant that detaches from the group (``setsid``/``setpgid``, as some
build tools do) cannot be reached here; see the module docstring.
:param process: The subprocess to terminate.
"""
if process.returncode is not None:
return
killpg = getattr(os, "killpg", None)
if killpg is None: # pragma: no cover - non-POSIX fallback
process.terminate()
try:
await asyncio.wait_for(process.wait(), timeout=KILL_GRACE_SECONDS)
except TimeoutError:
process.kill()
await process.wait()
return
# Capture the group id before signalling: once the leader is reaped,
# ``getpgid(pid)`` is no longer usable, but the group id stays valid while
# any member remains.
try:
pgid = os.getpgid(process.pid)
except (ProcessLookupError, PermissionError):
await process.wait()
return
try:
killpg(pgid, signal.SIGTERM)
except (ProcessLookupError, PermissionError):
await process.wait()
return
if await _drain_process_group(pgid, KILL_GRACE_SECONDS):
await process.wait()
return
logger.warning("Command process group did not exit after SIGTERM; sending SIGKILL")
try:
killpg(pgid, signal.SIGKILL)
except (ProcessLookupError, PermissionError):
pass
await process.wait()
[docs]
async def run_command(
command: str,
working_directory: str | None = None,
timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
max_output_chars: int = DEFAULT_MAX_OUTPUT_CHARS,
project_root: str | None = None,
) -> dict[str, Any]:
"""Run a shell command and return its exit status and captured output.
Framework-agnostic implementation shared across Klea MCP servers. Apps
wrap this in an MCP tool (see ``klea_utils.mcp.server.bundled_tools``).
This is **not** a sandbox (ADR-0038): the command can ignore
``working_directory`` (``cd /``, absolute paths) and has the host user's
filesystem, process, and network authority. The command inherits the
server's environment. Access control relies on the client-side path gate
(``working_directory`` and paths in the command, ADR-0007) plus the tool's
``destructive`` / ``open_world`` annotations and the access level
(ADR-0037).
The process is run with ``start_new_session`` so it can be killed as a
group on timeout, with stdin closed so an interactive command cannot
hang. Timeout and cancellation terminate the whole process group, but a
command that detaches its children (``setsid``/daemonising) can leave
processes running that the group signal cannot reach.
:param command: Shell command string to run.
:param working_directory: Directory to run in. Defaults to *project_root*
(or the current working directory when *project_root* is ``None``).
:param timeout_seconds: Seconds to wait before killing the process group.
Must not exceed :func:`max_timeout_seconds`.
:param max_output_chars: Maximum characters captured per stream.
:param project_root: Default base for *working_directory* when it is not
given; not a permission check (the client-side gate enforces the
boundary).
:returns: dict with command, working_directory, returncode, stdout,
stderr, truncated, error. A non-zero ``returncode`` is a normal
command result (many tools signal conditions with it); ``error`` is
set only when the command could not be run or was killed (timeout,
denied working directory, spawn failure).
"""
logger.debug(
f"Running command\n"
f"{command = }\n"
f"{working_directory = }\n"
f"{timeout_seconds = }\n"
f"{max_output_chars = }\n"
f"{project_root = }"
)
if not command.strip():
logger.warning("Empty command rejected")
return _result(command, None, error="Empty command is not allowed.")
if max_output_chars <= 0:
max_output_chars = DEFAULT_MAX_OUTPUT_CHARS
boundary = project_root
cwd = (
Path(working_directory)
if working_directory
else Path(boundary)
if boundary
else Path.cwd()
)
cwd = cwd.expanduser().resolve()
ceiling = max_timeout_seconds()
if timeout_seconds <= 0:
logger.warning(
f"Invalid timeout {timeout_seconds}; using {DEFAULT_TIMEOUT_SECONDS}"
)
timeout_seconds = DEFAULT_TIMEOUT_SECONDS
if timeout_seconds > ceiling:
logger.warning(f"Requested timeout {timeout_seconds} exceeds ceiling {ceiling}")
return _result(
command,
str(cwd),
error=(
f"timeout_seconds ({timeout_seconds:g}) exceeds the maximum "
f"({ceiling:g} s). Lower it or raise {MAX_TIMEOUT_ENV_VAR}."
),
)
# Validate the working directory and start the process under one
# ``OSError`` guard: ``is_dir()`` re-raises permission errors (only
# ENOENT/ENOTDIR/EBADF/ELOOP are swallowed), so an unreadable path must
# become a non-halting error like any other OS failure.
try:
if not cwd.is_dir():
logger.warning(f"Not a directory: {cwd}")
return _result(
command,
str(cwd),
error=f"Working directory is not a directory: {cwd}",
)
process = await asyncio.create_subprocess_shell(
command,
cwd=str(cwd),
stdin=asyncio.subprocess.DEVNULL,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
start_new_session=True,
)
except OSError as exc:
logger.warning(f"Could not use working directory or start command: {exc}")
return _result(command, str(cwd), error=f"Could not start command: {exc}")
try:
stdout_b, stderr_b = await asyncio.wait_for(
process.communicate(), timeout=timeout_seconds
)
except TimeoutError:
logger.warning(f"Command timed out after {timeout_seconds} s: {command}")
await _terminate_process_group(process)
return _result(
command,
str(cwd),
error=(
f"Command timed out after {timeout_seconds:g} s and was killed. "
"Retry with a larger timeout if it is expected to take longer."
),
)
except asyncio.CancelledError:
# The call was cancelled (client-side call timeout, graph cancel, or
# server shutdown). Shield the cleanup so the process group is still
# terminated before the cancellation propagates.
logger.warning(f"Command cancelled; terminating process group: {command}")
await asyncio.shield(_terminate_process_group(process))
raise
stdout = stdout_b.decode("utf-8", errors="replace")
stderr = stderr_b.decode("utf-8", errors="replace")
truncated = False
if len(stdout) > max_output_chars:
stdout = stdout[:max_output_chars]
truncated = True
if len(stderr) > max_output_chars:
stderr = stderr[:max_output_chars]
truncated = True
returncode = process.returncode
# A non-zero exit is a command *result*, not a tool failure: tools such as
# ``diff``/``grep``/``test`` use exit codes semantically. ``error`` (and
# hence MCP ``isError`` via ``to_result``) is reserved for the command
# never producing a result (timeout/denied/spawn), handled above; the
# caller judges the outcome from ``returncode``/``stdout``/``stderr``.
logger.debug(f"Command finished ({returncode}): {command}")
return _result(
command,
str(cwd),
returncode=returncode,
stdout=stdout,
stderr=stderr,
truncated=truncated,
error="",
)