Last active
April 24, 2026 09:39
-
-
Save ag88/99e46ed64d7227bdca5ba3ced9189d2a to your computer and use it in GitHub Desktop.
a linux command running and url fetch MCP server
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| # the MCP server normally runs at http://127.0.0.1:8000/mcp | |
| # install: | |
| # - recommend: use a python virtual environment e.g. python3 -m venv venv_dir | |
| # - pip3 install the following: | |
| # - pip3 install mcp requests | |
| # run: | |
| # python3 LinuxCmd_n_UrlFetch_MCP.py | |
| from mcp.server.fastmcp import FastMCP | |
| import subprocess | |
| import json | |
| import glob | |
| import asyncio | |
| import requests | |
| from typing import Optional | |
| import logging | |
| log = logging.getLogger(__name__) | |
| # Initialize the server | |
| #mcp = FastMCP(name="LimitedShell", log_level="DEBUG") | |
| mcp = FastMCP(name="LimitedShell", log_level="INFO") | |
| # Define your allowed whitelist | |
| ALLOWED_COMMANDS = ["ls", "pwd", "echo", "date", "whoami"] | |
| # more | |
| MORE_COMMANDS = ["cat", "head", "tail", "grep", "egrep", "sed", "wc", "file", "du", "df", "free", "ps", | |
| "uname", "hostname", "uptime", "w", "last"] | |
| ALLOWED_COMMANDS.extend(MORE_COMMANDS) | |
| @mcp.tool() | |
| def safe_shell(command: str, args: Optional[list[str]] = None) -> str: | |
| """ | |
| Runs any of the following linux/unix commands: | |
| ["ls", "pwd", "echo", "date", "whoami", "cat", "head", "tail", "grep", "egrep", "sed", | |
| "wc", "file", "du", "df", "free", "ps", "uname", "hostname", "uptime", "w", "last"] | |
| Args: | |
| command: The command name (e.g., 'ls') | |
| args: A list of command arguments/flags (e.g., ['-la', '*.txt']) | |
| use quotes e.g. ['"*.txt"'] for literals that should not | |
| be expanded into file names | |
| note: pipes are not allowed | |
| """ | |
| command = command.strip() | |
| if command.find(" ") > 0: | |
| return "Error: separate and place each option and argument in the args list/array" | |
| if command not in ALLOWED_COMMANDS: | |
| return f"Access Denied: '{command}' is not in the allowed list." | |
| if args is None: | |
| args = [] | |
| log.info("cmd: {} {}".format(command, " ".join(args))) | |
| # Resolve shell glob patterns in arguments | |
| expanded_args = [] | |
| for arg in args: | |
| # Check for common glob characters | |
| arg = arg.encode().decode('unicode_escape') | |
| if arg.startswith('"') and arg.endswith('"'): | |
| arg = arg[1:-1] | |
| expanded_args.append(arg) | |
| else: | |
| if any(char in arg for char in ('*', '?', '[', ']')): | |
| try: | |
| matches = glob.glob(arg) | |
| if matches: | |
| # Expand to all matching files/directories | |
| expanded_args.extend(matches) | |
| else: | |
| # Keep original pattern if no matches (mimics standard shell behavior) | |
| expanded_args.append(arg) | |
| except Exception: | |
| # Fallback to original argument on glob resolution error | |
| expanded_args.append(arg) | |
| else: | |
| expanded_args.append(arg) | |
| try: | |
| # Construct command safely as a list to prevent shell injection | |
| full_cmd = [command] + expanded_args | |
| log.debug("full cmd: {}".format(" ".join(full_cmd))) | |
| result = subprocess.run( | |
| full_cmd, | |
| capture_output=True, | |
| text=True, | |
| timeout=100 # Prevents hanging processes | |
| ) | |
| return result.stdout if result.returncode == 0 else result.stderr | |
| except subprocess.TimeoutExpired: | |
| return "Error: Command execution timed out." | |
| except Exception as e: | |
| return f"Error executing command: {str(e)}" | |
| @mcp.tool() | |
| def fetch(url: str) -> str: | |
| """ | |
| fetch a page from a url | |
| """ | |
| r = requests.get(url) | |
| return r.text | |
| async def listtools(): | |
| tools = await mcp.list_tools() | |
| for t in tools: | |
| log.info(t.name) | |
| log.info(t.title) | |
| log.info(t.description) | |
| if __name__ == "__main__": | |
| #logging.basicConfig(level=logging.DEBUG) | |
| logging.basicConfig(level=logging.INFO) | |
| #tools = mcp._tool_manager.list_tools() | |
| asyncio.run(listtools()) | |
| log.info("allowed commands: {}".format(ALLOWED_COMMANDS)) | |
| mcp.run(transport="streamable-http") | |
| #mcp.run(transport="sse") | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment