Coverage for agentos/mcp/builtin_servers.py: 28%

1""" 

2Built-in MCP Servers for AgentOS (v1.8.1). 

3 

4Pure Python implementations of common MCP tools. 8 servers, 32+ tools total. 

5 

6Servers: 

7 FilesystemServer (7) - Safe file I/O with path validation 

8 WebFetchServer (3) - HTTP client with content extraction 

9 MemoryServer (6) - Persistent knowledge graph 

10 SearchServer (4) - Web search via DuckDuckGo 

11 GitServer (4) - Git operations 

12 ShellServer (3) - Safe shell command execution 

13 CodeServer (3) - Python/JS code execution in sandbox 

14 TextServer (4) - Text manipulation & formatting 

15""" 

16 

17from __future__ import annotations 

18 

19import json 

20import os 

21import re 

22import hashlib 

23import shutil 

24import subprocess 

25import time 

26import urllib.parse 

27import urllib.request 

28import urllib.error 

29from datetime import datetime 

30from pathlib import Path 

31from typing import Any, Dict, List, Optional 

32 

33 

34# ── Filesystem MCP Server (7 tools) ───────── 

35 

36 

37class FilesystemServer: 

38 """MCP-compatible filesystem server with safe path validation.""" 

39 

40 NAME = "filesystem" 

41 VERSION = "1.0.0" 

42 

43 def __init__(self, allowed_paths: Optional[List[str]] = None): 

44 self._allowed_paths = [ 

45 Path(p).resolve() 

46 for p in (allowed_paths or [os.getcwd(), str(Path.home())]) 

47 ] 

48 for p in self._allowed_paths: 

49 p.mkdir(parents=True, exist_ok=True) 

50 

51 def _validate_path(self, path_str: str) -> Path: 

52 p = Path(path_str).expanduser().resolve() 

53 for allowed in self._allowed_paths: 

54 try: 

55 p.relative_to(allowed) 

56 return p 

57 except ValueError: 

58 continue 

59 raise ValueError(f"Path '{path_str}' outside allowed directories") 

60 

61 def get_tools(self) -> List[Dict[str, Any]]: 

62 return [ 

63 {"name": "read_file", "description": "Read contents of a text file", 

64 "inputSchema": {"type": "object", "properties": { 

65 "path": {"type": "string"}, "encoding": {"type": "string", "default": "utf-8"}}, 

66 "required": ["path"]}}, 

67 {"name": "write_file", "description": "Write text content to a file", 

68 "inputSchema": {"type": "object", "properties": { 

69 "path": {"type": "string"}, "content": {"type": "string"}, 

70 "encoding": {"type": "string", "default": "utf-8"}}, 

71 "required": ["path", "content"]}}, 

72 {"name": "list_directory", "description": "List directory contents with metadata", 

73 "inputSchema": {"type": "object", "properties": { 

74 "path": {"type": "string"}, "recursive": {"type": "boolean", "default": False}}, 

75 "required": ["path"]}}, 

76 {"name": "search_files", "description": "Search files by glob pattern", 

77 "inputSchema": {"type": "object", "properties": { 

78 "path": {"type": "string"}, "pattern": {"type": "string"}}, 

79 "required": ["path", "pattern"]}}, 

80 {"name": "get_file_info", "description": "Get file/directory metadata", 

81 "inputSchema": {"type": "object", "properties": { 

82 "path": {"type": "string"}}, "required": ["path"]}}, 

83 {"name": "create_directory", "description": "Create directory and parents", 

84 "inputSchema": {"type": "object", "properties": { 

85 "path": {"type": "string"}}, "required": ["path"]}}, 

86 {"name": "move_file", "description": "Move or rename a file/directory", 

87 "inputSchema": {"type": "object", "properties": { 

88 "source": {"type": "string"}, "destination": {"type": "string"}}, 

89 "required": ["source", "destination"]}}, 

90 ] 

91 

92 def call_tool(self, tool_name: str, arguments: Dict) -> Any: 

93 return getattr(self, f"_handle_{tool_name}")(**arguments) 

94 

95 def _handle_read_file(self, path: str, encoding: str = "utf-8") -> str: 

96 p = self._validate_path(path) 

97 if not p.is_file(): raise FileNotFoundError(f"Not found: {p}") 

98 return p.read_text(encoding=encoding) 

99 

100 def _handle_write_file(self, path: str, content: str, encoding: str = "utf-8") -> str: 

101 p = self._validate_path(path) 

102 p.parent.mkdir(parents=True, exist_ok=True) 

103 p.write_text(content, encoding=encoding) 

104 return f"Wrote {len(content)} bytes to {p}" 

105 

106 def _handle_list_directory(self, path: str, recursive: bool = False) -> List[Dict]: 

107 p = self._validate_path(path) 

108 if not p.is_dir(): raise NotADirectoryError(f"Not a directory: {p}") 

109 entries = [] 

110 for item in (p.rglob("*") if recursive else p.iterdir()): 

111 if item.name.startswith("."): continue 

112 s = item.stat() 

113 entries.append({"name": item.name, "path": str(item), "size": s.st_size, 

114 "is_dir": item.is_dir(), "modified": datetime.fromtimestamp(s.st_mtime).isoformat()}) 

115 return sorted(entries, key=lambda e: (not e["is_dir"], e["name"])) 

116 

117 def _handle_search_files(self, path: str, pattern: str) -> List[Dict]: 

118 p = self._validate_path(path) 

119 if not p.is_dir(): raise NotADirectoryError(f"Not a directory: {p}") 

120 return sorted([{"name": i.name, "path": str(i), "size": i.stat().st_size, "is_dir": i.is_dir()} 

121 for i in p.rglob(pattern) if not i.name.startswith(".")], key=lambda e: e["path"]) 

122 

123 def _handle_get_file_info(self, path: str) -> Dict: 

124 p = self._validate_path(path) 

125 if not p.exists(): raise FileNotFoundError(f"Not found: {p}") 

126 s = p.stat(); ext = p.suffix.lower() 

127 mime = {".txt": "text/plain", ".md": "text/markdown", ".py": "text/x-python", 

128 ".json": "application/json", ".html": "text/html", ".pdf": "application/pdf"} 

129 return {"name": p.name, "path": str(p), "size": s.st_size, "is_dir": p.is_dir(), 

130 "is_file": p.is_file(), "extension": ext, "mime_type": mime.get(ext, "application/octet-stream"), 

131 "created": datetime.fromtimestamp(s.st_ctime).isoformat(), 

132 "modified": datetime.fromtimestamp(s.st_mtime).isoformat(), 

133 "permissions": oct(s.st_mode)[-3:]} 

134 

135 def _handle_create_directory(self, path: str) -> str: 

136 p = self._validate_path(path) 

137 p.mkdir(parents=True, exist_ok=True) 

138 return f"Created: {p}" 

139 

140 def _handle_move_file(self, source: str, destination: str) -> str: 

141 src = self._validate_path(source); dst = self._validate_path(destination) 

142 if not src.exists(): raise FileNotFoundError(f"Not found: {src}") 

143 src.rename(dst) 

144 return f"Moved {src} -> {dst}" 

145 

146 

147# ── Web Fetch MCP Server (3 tools) ─────────── 

148 

149 

150class WebFetchServer: 

151 """MCP-compatible HTTP client with content extraction.""" 

152 

153 NAME = "webfetch" 

154 VERSION = "1.0.0" 

155 

156 def __init__(self, user_agent: str = "AgentOS-MCP/1.0", timeout: int = 30): 

157 self._ua = user_agent; self._timeout = timeout 

158 

159 def get_tools(self) -> List[Dict[str, Any]]: 

160 return [ 

161 {"name": "fetch_url", "description": "Fetch a web page and return cleaned text", 

162 "inputSchema": {"type": "object", "properties": { 

163 "url": {"type": "string"}, "max_length": {"type": "integer", "default": 50000}}, 

164 "required": ["url"]}}, 

165 {"name": "fetch_json", "description": "Fetch and parse JSON from URL", 

166 "inputSchema": {"type": "object", "properties": { 

167 "url": {"type": "string"}, "headers": {"type": "object"}}, 

168 "required": ["url"]}}, 

169 {"name": "check_url", "description": "HEAD request to verify URL accessibility", 

170 "inputSchema": {"type": "object", "properties": { 

171 "url": {"type": "string"}}, "required": ["url"]}}, 

172 ] 

173 

174 def call_tool(self, tool_name: str, arguments: Dict) -> Any: 

175 return getattr(self, f"_handle_{tool_name}")(**arguments) 

176 

177 @staticmethod 

178 def _validate_url(url: str) -> None: 

179 p = urllib.parse.urlparse(url) 

180 if p.scheme not in ("http", "https"): raise ValueError(f"Invalid scheme: {p.scheme}") 

181 

182 @staticmethod 

183 def _strip_html(html: str) -> str: 

184 html = re.sub(r'<script[^>]*>.*?</script>', '', html, flags=re.DOTALL | re.I) 

185 html = re.sub(r'<style[^>]*>.*?</style>', '', html, flags=re.DOTALL | re.I) 

186 text = re.sub(r'<[^>]+>', ' ', html) 

187 text = re.sub(r'\s+', ' ', text).strip() 

188 for e, c in [('&amp;','&'),('&lt;','<'),('&gt;','>'),('&quot;','"'),('&#39;',"'"),('&nbsp;',' ')]: 

189 text = text.replace(e, c) 

190 return text 

191 

192 def _handle_fetch_url(self, url: str, max_length: int = 50000) -> str: 

193 self._validate_url(url) 

194 req = urllib.request.Request(url, headers={"User-Agent": self._ua}) 

195 try: 

196 with urllib.request.urlopen(req, timeout=self._timeout) as r: 

197 ct = r.headers.get("Content-Type", "") 

198 enc = ct.split("charset=")[-1].split(";")[0].strip() if "charset=" in ct else "utf-8" 

199 txt = r.read().decode(enc, errors="replace") 

200 if "text/html" in ct: txt = self._strip_html(txt) 

201 return txt[:max_length] + (f"\n\n[Truncated at {max_length}]" if len(txt) > max_length else "") 

202 except urllib.error.HTTPError as e: return f"HTTP {e.code}: {e.reason}" 

203 except Exception as e: return f"Error: {e}" 

204 

205 def _handle_fetch_json(self, url: str, headers: Optional[Dict] = None) -> Any: 

206 self._validate_url(url) 

207 h = {"User-Agent": self._ua, "Accept": "application/json"} 

208 if headers: h.update(headers) 

209 try: 

210 with urllib.request.urlopen(urllib.request.Request(url, headers=h), timeout=self._timeout) as r: 

211 return json.loads(r.read()) 

212 except json.JSONDecodeError as e: return {"error": "Invalid JSON", "detail": str(e)} 

213 except Exception as e: return {"error": str(e)} 

214 

215 def _handle_check_url(self, url: str) -> Dict: 

216 self._validate_url(url) 

217 req = urllib.request.Request(url, headers={"User-Agent": self._ua}, method="HEAD") 

218 try: 

219 with urllib.request.urlopen(req, timeout=self._timeout) as r: 

220 return {"url": url, "status": r.status, "accessible": 200 <= r.status < 400, 

221 "content_type": r.headers.get("Content-Type",""), "content_length": r.headers.get("Content-Length","")} 

222 except urllib.error.HTTPError as e: return {"url": url, "status": e.code, "accessible": False, "reason": e.reason} 

223 except Exception as e: return {"url": url, "status": 0, "accessible": False, "error": str(e)} 

224 

225 

226# ── Memory / Knowledge Graph MCP (6 tools) ─── 

227 

228 

229class MemoryServer: 

230 """Persistent knowledge graph for agent memory.""" 

231 

232 NAME = "memory" 

233 VERSION = "1.0.0" 

234 

235 def __init__(self, storage_path: str = ""): 

236 self._path = Path(storage_path or str(Path.home() / ".agentos" / "memory" / "kg.json")) 

237 self._path.parent.mkdir(parents=True, exist_ok=True) 

238 self._entries: Dict[str, Dict] = {} 

239 if self._path.exists(): 

240 try: 

241 for e in json.loads(self._path.read_text()).get("entries", []): self._entries[e["id"]] = e 

242 except: pass 

243 

244 def _save(self): 

245 self._path.write_text(json.dumps({"version": "1.0", "updated_at": time.time(), 

246 "entries": list(self._entries.values())}, indent=2, ensure_ascii=False)) 

247 

248 def _make_id(self, content: str) -> str: 

249 return hashlib.sha256(content.encode()).hexdigest()[:16] 

250 

251 def get_tools(self) -> List[Dict[str, Any]]: 

252 return [ 

253 {"name": "store_memory", "description": "Store a fact/memory", 

254 "inputSchema": {"type": "object", "properties": { 

255 "content": {"type": "string"}, "category": {"type": "string", "default": "general"}, 

256 "tags": {"type": "array", "items": {"type": "string"}}, "metadata": {"type": "object"}}, 

257 "required": ["content"]}}, 

258 {"name": "retrieve_memory", "description": "Retrieve memory by ID", 

259 "inputSchema": {"type": "object", "properties": {"memory_id": {"type": "string"}}, "required": ["memory_id"]}}, 

260 {"name": "search_memory", "description": "Search memories by keyword/category/tags", 

261 "inputSchema": {"type": "object", "properties": { 

262 "query": {"type": "string"}, "category": {"type": "string"}, 

263 "tags": {"type": "array", "items": {"type": "string"}}, "limit": {"type": "integer", "default": 20}}}}, 

264 {"name": "list_categories", "description": "List all categories with counts", 

265 "inputSchema": {"type": "object", "properties": {}}}, 

266 {"name": "delete_memory", "description": "Delete a memory by ID", 

267 "inputSchema": {"type": "object", "properties": {"memory_id": {"type": "string"}}, "required": ["memory_id"]}}, 

268 {"name": "update_memory", "description": "Update memory content/metadata", 

269 "inputSchema": {"type": "object", "properties": { 

270 "memory_id": {"type": "string"}, "content": {"type": "string"}, 

271 "category": {"type": "string"}, "tags": {"type": "array", "items": {"type": "string"}}}, 

272 "required": ["memory_id"]}}, 

273 ] 

274 

275 def call_tool(self, tool_name: str, arguments: Dict) -> Any: 

276 return getattr(self, f"_handle_{tool_name}")(**arguments) 

277 

278 def _handle_store_memory(self, content: str, category: str = "general", 

279 tags: list = None, metadata: dict = None) -> Dict: 

280 mid = self._make_id(content) 

281 entry = self._entries.get(mid, {}) 

282 if entry: 

283 entry["updated_at"] = time.time() 

284 if tags: entry.setdefault("tags", []).extend(t for t in tags if t not in entry.get("tags", [])) 

285 if category != "general": entry["category"] = category 

286 self._save(); return {"id": mid, "action": "updated"} 

287 entry = {"id": mid, "content": content, "category": category, "tags": tags or [], 

288 "metadata": metadata or {}, "created_at": time.time(), "updated_at": time.time()} 

289 self._entries[mid] = entry; self._save() 

290 return {"id": mid, "action": "stored"} 

291 

292 def _handle_retrieve_memory(self, memory_id: str) -> Optional[Dict]: 

293 return self._entries.get(memory_id) 

294 

295 def _handle_search_memory(self, query: str = "", category: str = None, 

296 tags: list = None, limit: int = 20) -> List[Dict]: 

297 results = [] 

298 for e in self._entries.values(): 

299 if category and e.get("category") != category: continue 

300 if tags and not all(t in e.get("tags", []) for t in tags): continue 

301 if query and query.lower() not in e.get("content","").lower(): continue 

302 results.append({"id": e["id"], "content": e["content"], "category": e.get("category",""), 

303 "tags": e.get("tags",[]), "created_at": e.get("created_at",0)}) 

304 return sorted(results, key=lambda r: r["created_at"], reverse=True)[:limit] 

305 

306 def _handle_list_categories(self) -> Dict[str, int]: 

307 c: Dict[str, int] = {} 

308 for e in self._entries.values(): c[e.get("category","general")] = c.get(e.get("category","general"), 0) + 1 

309 return dict(sorted(c.items(), key=lambda x: -x[1])) 

310 

311 def _handle_delete_memory(self, memory_id: str) -> Dict: 

312 if memory_id in self._entries: del self._entries[memory_id]; self._save(); return {"deleted": True, "id": memory_id} 

313 return {"deleted": False, "id": memory_id, "reason": "not found"} 

314 

315 def _handle_update_memory(self, memory_id: str, content: str = None, category: str = None, tags: list = None) -> Dict: 

316 e = self._entries.get(memory_id) 

317 if not e: return {"updated": False, "reason": "not found"} 

318 if content is not None: e["content"] = content 

319 if category is not None: e["category"] = category 

320 if tags is not None: e["tags"] = tags 

321 e["updated_at"] = time.time(); self._save() 

322 return {"updated": True, "id": memory_id} 

323 

324 

325# ── Web Search Server (4 tools) ────────────── 

326 

327 

328class SearchServer: 

329 """MCP-compatible web search via DuckDuckGo + Google fallback.""" 

330 

331 NAME = "search" 

332 VERSION = "1.0.0" 

333 

334 def get_tools(self) -> List[Dict[str, Any]]: 

335 return [ 

336 {"name": "web_search", "description": "Search the web", 

337 "inputSchema": {"type": "object", "properties": { 

338 "query": {"type": "string"}, "max_results": {"type": "integer", "default": 10}}, 

339 "required": ["query"]}}, 

340 {"name": "news_search", "description": "Search for recent news", 

341 "inputSchema": {"type": "object", "properties": { 

342 "query": {"type": "string"}, "max_results": {"type": "integer", "default": 10}}, 

343 "required": ["query"]}}, 

344 {"name": "image_search", "description": "Search for images", 

345 "inputSchema": {"type": "object", "properties": { 

346 "query": {"type": "string"}, "max_results": {"type": "integer", "default": 10}}, 

347 "required": ["query"]}}, 

348 {"name": "suggest", "description": "Get search autocomplete suggestions", 

349 "inputSchema": {"type": "object", "properties": { 

350 "query": {"type": "string"}}, "required": ["query"]}}, 

351 ] 

352 

353 def call_tool(self, tool_name: str, arguments: Dict) -> Any: 

354 return getattr(self, f"_handle_{tool_name}")(**arguments) 

355 

356 def _handle_web_search(self, query: str, max_results: int = 10) -> List[Dict]: 

357 """Search via DuckDuckGo HTML (no API key needed).""" 

358 q = urllib.parse.quote_plus(query) 

359 url = f"https://html.duckduckgo.com/html/?q={q}" 

360 req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) 

361 try: 

362 with urllib.request.urlopen(req, timeout=15) as r: 

363 html = r.read().decode("utf-8", errors="replace") 

364 except Exception as e: 

365 return [{"error": f"Search failed: {e}"}] 

366 

367 results = [] 

368 # Parse DuckDuckGo HTML results 

369 for m in re.finditer(r'<a[^>]*class="result__a"[^>]*href="([^"]*)"[^>]*>(.*?)</a>', html, re.DOTALL): 

370 if len(results) >= max_results: break 

371 link = m.group(1) 

372 title = re.sub(r'<[^>]+>', '', m.group(2)).strip() 

373 # Find snippet 

374 snippet = "" 

375 sn_match = re.search(r'<a[^>]*class="result__snippet"[^>]*>(.*?)</a>', html[m.end():m.end()+500], re.DOTALL) 

376 if sn_match: snippet = re.sub(r'<[^>]+>', '', sn_match.group(1)).strip() 

377 results.append({"title": title, "url": link, "snippet": snippet, "source": "duckduckgo"}) 

378 return results 

379 

380 def _handle_news_search(self, query: str, max_results: int = 10) -> List[Dict]: 

381 q = urllib.parse.quote_plus(f"{query} news") 

382 return self._handle_web_search(q, max_results) 

383 

384 def _handle_image_search(self, query: str, max_results: int = 10) -> List[Dict]: 

385 q = urllib.parse.quote_plus(query) 

386 url = f"https://duckduckgo.com/?q={q}&iax=images&ia=images" 

387 req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) 

388 try: 

389 with urllib.request.urlopen(req, timeout=15) as r: 

390 html = r.read().decode("utf-8", errors="replace") 

391 except Exception as e: 

392 return [{"error": str(e)}] 

393 

394 # Extract vqd token for image search 

395 vqd = "" 

396 vqd_m = re.search(r'vqd=([\d-]+)', html) 

397 if vqd_m: vqd = vqd_m.group(1) 

398 

399 if vqd: 

400 img_url = f"https://duckduckgo.com/i.js?q={q}&vqd={vqd}&o=json&p=1&s=0" 

401 try: 

402 with urllib.request.urlopen(urllib.request.Request(img_url, headers={"User-Agent": "Mozilla/5.0"}), timeout=15) as r: 

403 data = json.loads(r.read()) 

404 results = data.get("results", [])[:max_results] 

405 return [{"title": r.get("title",""), "url": r.get("url",""), 

406 "thumbnail": r.get("thumbnail",""), "source": "duckduckgo"} for r in results] 

407 except: pass 

408 return [] 

409 

410 def _handle_suggest(self, query: str) -> List[str]: 

411 q = urllib.parse.quote_plus(query) 

412 url = f"https://duckduckgo.com/ac/?q={q}&type=list" 

413 try: 

414 with urllib.request.urlopen(url, timeout=8) as r: 

415 data = json.loads(r.read()) 

416 return [item.get("phrase","") for item in data[:10]] 

417 except: return [] 

418 

419 

420# ── Git Server (4 tools) ───────────────────── 

421 

422 

423class GitServer: 

424 """MCP-compatible git operations.""" 

425 

426 NAME = "git" 

427 VERSION = "1.0.0" 

428 

429 def __init__(self, repo_path: str = ""): 

430 self._repo_path = Path(repo_path).resolve() if repo_path else Path.cwd() 

431 self._git: Optional[str] = shutil.which("git") 

432 

433 def _run_git(self, *args) -> Dict[str, Any]: 

434 if not self._git: return {"error": "git not installed"} 

435 try: 

436 r = subprocess.run([self._git, *args], capture_output=True, text=True, 

437 cwd=str(self._repo_path), timeout=30) 

438 return {"stdout": r.stdout.strip(), "stderr": r.stderr.strip(), "returncode": r.returncode} 

439 except subprocess.TimeoutExpired: 

440 return {"error": "Command timed out"} 

441 except Exception as e: 

442 return {"error": str(e)} 

443 

444 def get_tools(self) -> List[Dict[str, Any]]: 

445 return [ 

446 {"name": "git_status", "description": "Show working tree status", 

447 "inputSchema": {"type": "object", "properties": {}}}, 

448 {"name": "git_log", "description": "Show commit history", 

449 "inputSchema": {"type": "object", "properties": { 

450 "max_count": {"type": "integer", "default": 10}, "oneline": {"type": "boolean", "default": True}}}}, 

451 {"name": "git_diff", "description": "Show changes between commits/working tree", 

452 "inputSchema": {"type": "object", "properties": { 

453 "staged": {"type": "boolean", "default": False}, "commit": {"type": "string"}}}}, 

454 {"name": "git_branch", "description": "List branches", 

455 "inputSchema": {"type": "object", "properties": {"remote": {"type": "boolean", "default": False}}}}, 

456 ] 

457 

458 def call_tool(self, tool_name: str, arguments: Dict) -> Any: 

459 return getattr(self, f"_handle_{tool_name}")(**arguments) 

460 

461 def _handle_git_status(self) -> Dict: 

462 return self._run_git("status", "--porcelain") 

463 

464 def _handle_git_log(self, max_count: int = 10, oneline: bool = True) -> Dict: 

465 args = ["log", f"-n{max_count}"] 

466 if oneline: args.append("--oneline") 

467 return self._run_git(*args) 

468 

469 def _handle_git_diff(self, staged: bool = False, commit: str = "") -> Dict: 

470 args = ["diff"] 

471 if staged: args.append("--staged") 

472 if commit: args.append(commit) 

473 return self._run_git(*args) 

474 

475 def _handle_git_branch(self, remote: bool = False) -> Dict: 

476 args = ["branch"] 

477 if remote: args.append("-r") 

478 return self._run_git(*args) 

479 

480 

481# ── Shell Server (3 tools) ─────────────────── 

482 

483 

484class ShellServer: 

485 """MCP-compatible safe shell command execution.""" 

486 

487 NAME = "shell" 

488 VERSION = "1.0.0" 

489 

490 SAFE_COMMANDS = {"ls", "cat", "head", "tail", "wc", "grep", "find", "du", "df", 

491 "echo", "date", "whoami", "uname", "pwd", "which", "env", "ps", 

492 "top", "htop", "tree", "file", "stat", "md5sum", "sha256sum", 

493 "python3", "python", "pip", "npm", "node", "curl", "wget"} 

494 

495 def _is_safe(self, cmd: str) -> bool: 

496 base = cmd.strip().split()[0] if cmd.strip() else "" 

497 return base in self.SAFE_COMMANDS 

498 

499 def get_tools(self) -> List[Dict[str, Any]]: 

500 return [ 

501 {"name": "run_command", "description": "Execute a safe shell command", 

502 "inputSchema": {"type": "object", "properties": { 

503 "command": {"type": "string"}, "timeout": {"type": "integer", "default": 30}}, 

504 "required": ["command"]}}, 

505 {"name": "system_info", "description": "Get system information (OS, CPU, memory)", 

506 "inputSchema": {"type": "object", "properties": {}}}, 

507 {"name": "disk_usage", "description": "Show disk usage for a path", 

508 "inputSchema": {"type": "object", "properties": { 

509 "path": {"type": "string", "default": "."}}}}, 

510 ] 

511 

512 def call_tool(self, tool_name: str, arguments: Dict) -> Any: 

513 return getattr(self, f"_handle_{tool_name}")(**arguments) 

514 

515 def _handle_run_command(self, command: str, timeout: int = 30) -> Dict: 

516 if not self._is_safe(command): 

517 return {"error": f"Command not in safelist. Allowed: {sorted(self.SAFE_COMMANDS)}"} 

518 try: 

519 r = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=timeout) 

520 return {"stdout": r.stdout, "stderr": r.stderr, "returncode": r.returncode} 

521 except subprocess.TimeoutExpired: return {"error": "Timeout"} 

522 except Exception as e: return {"error": str(e)} 

523 

524 def _handle_system_info(self) -> Dict: 

525 import platform 

526 return {"os": platform.system(), "release": platform.release(), "version": platform.version(), 

527 "machine": platform.machine(), "processor": platform.processor(), 

528 "python": platform.python_version(), "hostname": platform.node()} 

529 

530 def _handle_disk_usage(self, path: str = ".") -> Dict: 

531 try: 

532 usage = shutil.disk_usage(path) 

533 return {"path": path, "total_gb": round(usage.total/1024**3, 2), 

534 "used_gb": round(usage.used/1024**3, 2), "free_gb": round(usage.free/1024**3, 2)} 

535 except Exception as e: return {"error": str(e)} 

536 

537 

538# ── Code Server (3 tools) ──────────────────── 

539 

540 

541class CodeServer: 

542 """MCP-compatible sandboxed code execution.""" 

543 

544 NAME = "code" 

545 VERSION = "1.0.0" 

546 

547 def get_tools(self) -> List[Dict[str, Any]]: 

548 return [ 

549 {"name": "run_python", "description": "Execute Python code in sandbox", 

550 "inputSchema": {"type": "object", "properties": { 

551 "code": {"type": "string"}, "timeout": {"type": "integer", "default": 10}}, 

552 "required": ["code"]}}, 

553 {"name": "run_shell", "description": "Execute a one-liner bash command",