server.py (4454 bytes)
1 # /// script 2 # requires-python = ">=3.10" 3 # dependencies = ["mcp>=2,<3"] 4 # /// 5 """MCP server: a registry of DSLs. 6 7 Each DSL is a folder in $DSL_HOME (default ~/.dsl) that holds a grammar and 8 one interpreter file, impl.<ext>. `run` pipes a program to that interpreter. 9 """ 10 11 import json 12 import os 13 import re 14 import shutil 15 import subprocess 16 import sys 17 from pathlib import Path 18 19 from mcp.server.mcpserver import MCPServer 20 21 # The client sees the message of a ToolError. For any other exception it sees 22 # only "Error executing tool <name>". 23 from mcp.server.mcpserver.exceptions import ToolError 24 25 mcp = MCPServer("dsl") 26 27 NAME = re.compile(r"[a-z0-9][a-z0-9_-]{0,63}") 28 LIMIT = 100_000 # characters of stdout, and of stderr, that `run` returns 29 30 # Interpreter argv for each impl extension. `run` appends the impl path. 31 RUNTIMES: dict[str, list[str]] = { 32 "awk": ["awk", "-f"], 33 "sh": ["sh"], 34 "lisp": ["sbcl", "--script"], 35 "scm": ["guile", "--no-auto-compile", "-s"], 36 "hs": ["runghc"], 37 "clj": ["bb"], 38 "js": ["node"], 39 "py": [sys.executable], 40 } 41 42 43 def home() -> Path: 44 return Path(os.environ.get("DSL_HOME") or Path.home() / ".dsl") 45 46 47 def table() -> dict[str, list[str]]: 48 """RUNTIMES, plus the entries of $DSL_HOME/runtimes.json when it exists.""" 49 override = home() / "runtimes.json" 50 extra: dict[str, list[str]] = ( 51 json.loads(override.read_bytes()) if override.exists() else {} 52 ) 53 return RUNTIMES | extra 54 55 56 def folder(name: str) -> Path: 57 # The name becomes a path segment, so reject anything that can leave home(). 58 if not NAME.fullmatch(name): 59 raise ToolError(f"name must match {NAME.pattern}") 60 return home() / name 61 62 63 def impl(name: str) -> Path: 64 found = sorted(folder(name).glob("impl.*")) 65 if not found: 66 raise ToolError(f"no DSL named {name}") 67 return found[0] 68 69 70 @mcp.tool() 71 def runtimes() -> dict[str, bool]: 72 """Map each impl extension to whether its interpreter is on PATH.""" 73 return {ext: shutil.which(argv[0]) is not None for ext, argv in table().items()} 74 75 76 @mcp.tool() 77 def ls() -> dict[str, str]: 78 """Map each stored DSL name to the extension of its interpreter.""" 79 return {f.parent.name: f.suffix[1:] for f in sorted(home().glob("*/impl.*"))} 80 81 82 @mcp.tool() 83 def get(name: str) -> dict[str, str]: 84 """Return the grammar, the extension and the interpreter source of a DSL.""" 85 f = impl(name) 86 return { 87 "grammar": (f.parent / "grammar").read_bytes().decode(), 88 "ext": f.suffix[1:], 89 "source": f.read_bytes().decode(), 90 } 91 92 93 @mcp.tool() 94 def put(name: str, grammar: str, ext: str, source: str) -> str: 95 """Store or replace a DSL. 96 97 `source` is the interpreter, in the language of `ext` (a key of `runtimes`). 98 It reads a program on stdin, writes the result to stdout, and exits non-zero 99 with a message on stderr for a bad program. 100 """ 101 if ext not in table(): 102 raise ToolError(f"ext must be one of {sorted(table())}") 103 d = folder(name) 104 d.mkdir(parents=True, exist_ok=True) 105 new = d / f"impl.{ext}" 106 (d / "grammar").write_bytes(grammar.encode()) 107 new.write_bytes(source.encode()) 108 for old in d.glob("impl.*"): 109 if old != new: 110 old.unlink() 111 return f"{name}.{ext}" 112 113 114 @mcp.tool() 115 def run(name: str, program: str, timeout: float = 10) -> dict[str, str | int]: 116 """Run a program through the interpreter of a DSL. 117 118 The interpreter starts in the working directory of the server. 119 Return stdout, stderr and the exit code. 120 """ 121 f = impl(name) 122 argv = table().get(f.suffix[1:]) 123 if argv is None: 124 raise ToolError(f"no runtime for {f.name}") 125 exe = shutil.which(argv[0]) 126 if exe is None: 127 raise ToolError(f"{argv[0]} is not on PATH") 128 # ponytail: captures all output in memory, then cuts it to LIMIT. 129 # Read through a capped pipe if a DSL can flood stdout within the timeout. 130 try: 131 done = subprocess.run( 132 [exe, *argv[1:], str(f)], 133 input=program.encode(), 134 capture_output=True, 135 timeout=timeout, 136 check=False, 137 ) 138 except subprocess.TimeoutExpired: 139 raise ToolError(f"timed out after {timeout} s") from None 140 return { 141 "stdout": done.stdout.decode(errors="replace")[:LIMIT], 142 "stderr": done.stderr.decode(errors="replace")[:LIMIT], 143 "exit": done.returncode, 144 } 145 146 147 if __name__ == "__main__": 148 mcp.run()