execute
Execute shell commands on remote SSH servers to manage systems, run scripts, and perform administrative tasks through configured server connections.
Instructions
Execute a shell command on a single SSH server.
Args: server: Server name (e.g. 'pro-dicentra', 'inf-ai'). Must match a configured server. Use list_servers to see available servers. command: Shell command to execute on the remote server. timeout: Command timeout in seconds. Default 30. working_dir: Remote directory to execute from. Uses server default if omitted. force: Bypass dangerous command detection. Use with extreme caution. Default false.
Returns: Formatted command execution result with stdout, stderr, and exit code.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| server | Yes | ||
| command | Yes | ||
| timeout | No | ||
| working_dir | No | ||
| force | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/ssh_mcp/server.py:180-209 (handler)The execute tool handler decorated with @mcp.tool(). This is the MCP tool entry point that initializes the server, calls the SSHManager.execute method, and formats the result.
@mcp.tool() async def execute( server: str, command: str, timeout: int = 30, working_dir: str | None = None, force: bool = False, ) -> str: """Execute a shell command on a single SSH server. Args: server: Server name (e.g. 'pro-dicentra', 'inf-ai'). Must match a configured server. Use list_servers to see available servers. command: Shell command to execute on the remote server. timeout: Command timeout in seconds. Default 30. working_dir: Remote directory to execute from. Uses server default if omitted. force: Bypass dangerous command detection. Use with extreme caution. Default false. Returns: Formatted command execution result with stdout, stderr, and exit code. """ try: _init() result = await _ssh.execute(server, command, timeout, working_dir, force) return format_exec_result(result) except Exception as e: logger.error(f"Error executing command on {server}: {e}") return f"Error executing command on {server}: {e}" - src/ssh_mcp/ssh.py:122-273 (handler)The core implementation of the execute method in SSHManager class. Handles dangerous command detection, SSH connection management, command execution with timeout, output truncation, and comprehensive error handling.
async def execute( self, server_name: str, command: str, timeout: int = 30, working_dir: str | None = None, force: bool = False, ) -> ExecResult: """Execute command on a remote server. Args: server_name: Server name from registry command: Command to execute timeout: Command timeout in seconds working_dir: Working directory for command execution force: Bypass dangerous command detection (use with caution) Returns: ExecResult with command output and metadata """ try: server = self.registry.get_server(server_name) # Check for dangerous commands unless force is enabled if not force and _is_dangerous_command(command): logger.warning( f"Blocked potentially destructive command on {server_name}: {command}" ) return ExecResult( server=server_name, command=command, stdout="", stderr="", exit_code=None, error="Blocked: potentially destructive command detected. Review and use with caution.", ) # Use server-specific timeout if configured effective_timeout = server.timeout or timeout # Prepend working directory if specified effective_command = command if working_dir: effective_command = f"cd {shlex.quote(working_dir)} && {command}" elif server.default_dir: effective_command = f"cd {shlex.quote(server.default_dir)} && {command}" # Get or create connection conn = await self._get_connection(server_name) # Execute command and track duration start_time = time.monotonic() try: result = await conn.run(effective_command, timeout=effective_timeout) duration_ms = int((time.monotonic() - start_time) * 1000) # Truncate output if needed stdout = result.stdout or "" stderr = result.stderr or "" if len(stdout) > self.settings.max_output_bytes: stdout = ( stdout[: self.settings.max_output_bytes] + f"\n[... output truncated at {self.settings.max_output_bytes} bytes]" ) if len(stderr) > self.settings.max_output_bytes: stderr = ( stderr[: self.settings.max_output_bytes] + f"\n[... output truncated at {self.settings.max_output_bytes} bytes]" ) exec_result = ExecResult( server=server_name, command=command, stdout=stdout, stderr=stderr, exit_code=result.exit_status, duration_ms=duration_ms, ) # Audit log successful execution self._audit.info( "server=%s command=%s exit_code=%s duration_ms=%s", server_name, command, exec_result.exit_code, exec_result.duration_ms, ) return exec_result except asyncio.TimeoutError as e: duration_ms = int((time.monotonic() - start_time) * 1000) logger.error(f"Command timeout on {server_name}: {command}") exec_result = ExecResult( server=server_name, command=command, stdout="", stderr="", exit_code=None, error=f"Command timeout after {effective_timeout}s: {e}", duration_ms=duration_ms, ) # Audit log timeout self._audit.info( "server=%s command=%s exit_code=%s duration_ms=%s error=timeout", server_name, command, exec_result.exit_code, exec_result.duration_ms, ) return exec_result except KeyError as e: logger.error(f"Server not found: {server_name}") return ExecResult( server=server_name, command=command, stdout="", stderr="", exit_code=None, error=f"Server not found: {e}", ) except ( asyncssh.DisconnectError, asyncssh.PermissionDenied, OSError, ) as e: logger.error(f"SSH error on {server_name}: {e}") return ExecResult( server=server_name, command=command, stdout="", stderr="", exit_code=None, error=f"SSH error: {e}", ) except Exception as e: logger.error(f"Unexpected error on {server_name}: {e}") return ExecResult( server=server_name, command=command, stdout="", stderr="", exit_code=None, error=f"Unexpected error: {e}", ) - src/ssh_mcp/models.py:79-100 (schema)ExecResult dataclass defining the result structure returned by the execute tool, containing server, command, stdout, stderr, exit_code, error, and duration_ms fields.
class ExecResult: """Result from executing a command on a remote server. Mutable to allow construction during execution. Attributes: server: Server name where command was executed command: The command that was executed stdout: Standard output captured from command stderr: Standard error captured from command exit_code: Process exit code (None if execution failed) error: Error message if execution failed duration_ms: Command execution duration in milliseconds """ server: str command: str stdout: str stderr: str exit_code: int | None error: str | None = None duration_ms: int = 0 - src/ssh_mcp/server.py:180-180 (registration)The @mcp.tool() decorator registers the execute function as an MCP tool with the FastMCP server instance.
@mcp.tool()