Coverage for agentos/mcp/builtin

1"""

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

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

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"""

17from __future__ import annotations

19import hashlib

20import json

21import os

22import re

23import shutil

24import subprocess

25import time

26import urllib.error

27import urllib.parse

28import urllib.request

29from datetime import datetime

30from pathlib import Path

31from typing import Any

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

36class FilesystemServer:

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

39 NAME = "filesystem"

40 VERSION = "1.0.0"

42 def __init__(self, allowed_paths: list[str] | None = None):

43 self._allowed_paths = [

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

45 ]

46 for p in self._allowed_paths:

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

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

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

51 for allowed in self._allowed_paths:

52 try:

53 p.relative_to(allowed)

54 return p

55 except ValueError:

56 continue

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

59 def get_tools(self) -> list[dict[str, Any]]:

60 return [

61 {

62 "name": "read_file",

63 "description": "Read contents of a text file",

64 "inputSchema": {

65 "type": "object",

66 "properties": {

67 "path": {"type": "string"},

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

69 },

70 "required": ["path"],

71 },

72 },

73 {

74 "name": "write_file",

75 "description": "Write text content to a file",

76 "inputSchema": {

77 "type": "object",

78 "properties": {

79 "path": {"type": "string"},

80 "content": {"type": "string"},

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

82 },

83 "required": ["path", "content"],

84 },

85 },

86 {

87 "name": "list_directory",

88 "description": "List directory contents with metadata",

89 "inputSchema": {

90 "type": "object",

91 "properties": {

92 "path": {"type": "string"},

93 "recursive": {"type": "boolean", "default": False},

94 },

95 "required": ["path"],

96 },

97 },

98 {

99 "name": "search_files",

100 "description": "Search files by glob pattern",

101 "inputSchema": {

102 "type": "object",

103 "properties": {"path": {"type": "string"}, "pattern": {"type": "string"}},

104 "required": ["path", "pattern"],

105 },

106 },

107 {

108 "name": "get_file_info",

109 "description": "Get file/directory metadata",

110 "inputSchema": {

111 "type": "object",

112 "properties": {"path": {"type": "string"}},

113 "required": ["path"],

114 },

115 },

116 {

117 "name": "create_directory",

118 "description": "Create directory and parents",

119 "inputSchema": {

120 "type": "object",

121 "properties": {"path": {"type": "string"}},

122 "required": ["path"],

123 },

124 },

125 {

126 "name": "move_file",

127 "description": "Move or rename a file/directory",

128 "inputSchema": {

129 "type": "object",

130 "properties": {"source": {"type": "string"}, "destination": {"type": "string"}},

131 "required": ["source", "destination"],

132 },

133 },

134 ]

135

136 def call_tool(self, tool_name: str, arguments: dict) -> Any:

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

138

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

140 p = self._validate_path(path)

141 if not p.is_file():

142 raise FileNotFoundError(f"Not found: {p}")

143 return p.read_text(encoding=encoding)

144

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

146 p = self._validate_path(path)

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

148 p.write_text(content, encoding=encoding)

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

150

151 def _handle_list_directory(self, path: str, recursive: bool = False) -> list[dict]:

152 p = self._validate_path(path)

153 if not p.is_dir():

154 raise NotADirectoryError(f"Not a directory: {p}")

155 entries = []

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

157 if item.name.startswith("."):

158 continue

159 s = item.stat()

160 entries.append(

161 {

162 "name": item.name,

163 "path": str(item),

164 "size": s.st_size,

165 "is_dir": item.is_dir(),

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

167 }

168 )

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

170

171 def _handle_search_files(self, path: str, pattern: str) -> list[dict]:

172 p = self._validate_path(path)

173 if not p.is_dir():

174 raise NotADirectoryError(f"Not a directory: {p}")

175 return sorted(

176 [

177 {"name": i.name, "path": str(i), "size": i.stat().st_size, "is_dir": i.is_dir()}

178 for i in p.rglob(pattern)

179 if not i.name.startswith(".")

180 ],

181 key=lambda e: e["path"],

182 )

183

184 def _handle_get_file_info(self, path: str) -> dict:

185 p = self._validate_path(path)

186 if not p.exists():

187 raise FileNotFoundError(f"Not found: {p}")

188 s = p.stat()

189 ext = p.suffix.lower()

190 mime = {

191 ".txt": "text/plain",

192 ".md": "text/markdown",

193 ".py": "text/x-python",

194 ".json": "application/json",

195 ".html": "text/html",

196 ".pdf": "application/pdf",

197 }

198 return {

199 "name": p.name,

200 "path": str(p),

201 "size": s.st_size,

202 "is_dir": p.is_dir(),

203 "is_file": p.is_file(),

204 "extension": ext,

205 "mime_type": mime.get(ext, "application/octet-stream"),

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

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

208 "permissions": oct(s.st_mode)[-3:],

209 }

210

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

212 p = self._validate_path(path)

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

214 return f"Created: {p}"

215

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

217 src = self._validate_path(source)

218 dst = self._validate_path(destination)

219 if not src.exists():

220 raise FileNotFoundError(f"Not found: {src}")

221 src.rename(dst)

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

223

224

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

226

227

228class WebFetchServer:

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

230

231 NAME = "webfetch"

232 VERSION = "1.0.0"

233

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

235 self._ua = user_agent

236 self._timeout = timeout

237

238 def get_tools(self) -> list[dict[str, Any]]:

239 return [

240 {

241 "name": "fetch_url",

242 "description": "Fetch a web page and return cleaned text",

243 "inputSchema": {

244 "type": "object",

245 "properties": {

246 "url": {"type": "string"},

247 "max_length": {"type": "integer", "default": 50000},

248 },

249 "required": ["url"],

250 },

251 },

252 {

253 "name": "fetch_json",

254 "description": "Fetch and parse JSON from URL",

255 "inputSchema": {

256 "type": "object",

257 "properties": {"url": {"type": "string"}, "headers": {"type": "object"}},

258 "required": ["url"],

259 },

260 },

261 {

262 "name": "check_url",

263 "description": "HEAD request to verify URL accessibility",

264 "inputSchema": {

265 "type": "object",

266 "properties": {"url": {"type": "string"}},

267 "required": ["url"],

268 },

269 },

270 ]

271

272 def call_tool(self, tool_name: str, arguments: dict) -> Any:

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

274

275 @staticmethod

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

277 p = urllib.parse.urlparse(url)

278 if p.scheme not in ("http", "https"):

279 raise ValueError(f"Invalid scheme: {p.scheme}")

280

281 @staticmethod

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

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

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

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

286 text = re.sub(r"\s+", " ", text).strip()

287 for e, c in [

288 ("&", "&"),

289 ("<", "<"),

290 (">", ">"),

291 (""", '"'),

292 ("'", "'"),

293 (" ", " "),

294 ]:

295 text = text.replace(e, c)

296 return text

297

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

299 self._validate_url(url)

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

301 try:

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

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

304 enc = (

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

306 )

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

308 if "text/html" in ct:

309 txt = self._strip_html(txt)

310 return txt[:max_length] + (

311 f"\n\n[Truncated at {max_length}]" if len(txt) > max_length else ""

312 )

313 except urllib.error.HTTPError as e:

314 return f"HTTP {e.code}: {e.reason}"

315 except Exception as e:

316 return f"Error: {e}"

317

318 def _handle_fetch_json(self, url: str, headers: dict | None = None) -> Any:

319 self._validate_url(url)

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

321 if headers:

322 h.update(headers)

323 try:

324 with urllib.request.urlopen(

325 urllib.request.Request(url, headers=h), timeout=self._timeout

326 ) as r:

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

328 except json.JSONDecodeError as e:

329 return {"error": "Invalid JSON", "detail": str(e)}

330 except Exception as e:

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

332

333 def _handle_check_url(self, url: str) -> dict:

334 self._validate_url(url)

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

336 try:

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

338 return {

339 "url": url,

340 "status": r.status,

341 "accessible": 200 <= r.status < 400,

342 "content_type": r.headers.get("Content-Type", ""),

343 "content_length": r.headers.get("Content-Length", ""),

344 }

345 except urllib.error.HTTPError as e:

346 return {"url": url, "status": e.code, "accessible": False, "reason": e.reason}

347 except Exception as e:

348 return {"url": url, "status": 0, "accessible": False, "error": str(e)}

349

350

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

352

353

354class MemoryServer:

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

356

357 NAME = "memory"

358 VERSION = "1.0.0"

359

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

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

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

363 self._entries: dict[str, dict] = {}

364 if self._path.exists():

365 try:

366 for e in json.loads(self._path.read_text()).get("entries", []):

367 self._entries[e["id"]] = e

368 except:

369 pass

370

371 def _save(self):

372 self._path.write_text(

373 json.dumps(

374 {

375 "version": "1.0",

376 "updated_at": time.time(),

377 "entries": list(self._entries.values()),

378 },

379 indent=2,

380 ensure_ascii=False,

381 )

382 )

383

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

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

386

387 def get_tools(self) -> list[dict[str, Any]]:

388 return [

389 {

390 "name": "store_memory",

391 "description": "Store a fact/memory",

392 "inputSchema": {

393 "type": "object",

394 "properties": {

395 "content": {"type": "string"},

396 "category": {"type": "string", "default": "general"},

397 "tags": {"type": "array", "items": {"type": "string"}},

398 "metadata": {"type": "object"},

399 },

400 "required": ["content"],

401 },

402 },

403 {

404 "name": "retrieve_memory",

405 "description": "Retrieve memory by ID",

406 "inputSchema": {

407 "type": "object",

408 "properties": {"memory_id": {"type": "string"}},

409 "required": ["memory_id"],

410 },

411 },

412 {

413 "name": "search_memory",

414 "description": "Search memories by keyword/category/tags",

415 "inputSchema": {

416 "type": "object",

417 "properties": {

418 "query": {"type": "string"},

419 "category": {"type": "string"},

420 "tags": {"type": "array", "items": {"type": "string"}},

421 "limit": {"type": "integer", "default": 20},

422 },

423 },

424 },

425 {

426 "name": "list_categories",

427 "description": "List all categories with counts",

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

429 },

430 {

431 "name": "delete_memory",

432 "description": "Delete a memory by ID",

433 "inputSchema": {

434 "type": "object",

435 "properties": {"memory_id": {"type": "string"}},

436 "required": ["memory_id"],

437 },

438 },

439 {

440 "name": "update_memory",

441 "description": "Update memory content/metadata",

442 "inputSchema": {

443 "type": "object",

444 "properties": {

445 "memory_id": {"type": "string"},

446 "content": {"type": "string"},

447 "category": {"type": "string"},

448 "tags": {"type": "array", "items": {"type": "string"}},

449 },

450 "required": ["memory_id"],

451 },

452 },

453 ]

454

455 def call_tool(self, tool_name: str, arguments: dict) -> Any:

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

457

458 def _handle_store_memory(

459 self, content: str, category: str = "general", tags: list = None, metadata: dict = None

460 ) -> dict:

461 mid = self._make_id(content)

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

463 if entry:

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

465 if tags:

466 entry.setdefault("tags", []).extend(

467 t for t in tags if t not in entry.get("tags", [])

468 )

469 if category != "general":

470 entry["category"] = category

471 self._save()

472 return {"id": mid, "action": "updated"}

473 entry = {

474 "id": mid,

475 "content": content,

476 "category": category,

477 "tags": tags or [],

478 "metadata": metadata or {},

479 "created_at": time.time(),

480 "updated_at": time.time(),

481 }

482 self._entries[mid] = entry

483 self._save()

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

485

486 def _handle_retrieve_memory(self, memory_id: str) -> dict | None:

487 return self._entries.get(memory_id)

488

489 def _handle_search_memory(

490 self, query: str = "", category: str = None, tags: list = None, limit: int = 20

491 ) -> list[dict]:

492 results = []

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

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

495 continue

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

497 continue

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

499 continue

500 results.append(

501 {

502 "id": e["id"],

503 "content": e["content"],

504 "category": e.get("category", ""),

505 "tags": e.get("tags", []),

506 "created_at": e.get("created_at", 0),

507 }

508 )

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

510

511 def _handle_list_categories(self) -> dict[str, int]:

512 c: dict[str, int] = {}

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

514 c[e.get("category", "general")] = c.get(e.get("category", "general"), 0) + 1

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

516

517 def _handle_delete_memory(self, memory_id: str) -> dict:

518 if memory_id in self._entries:

519 del self._entries[memory_id]

520 self._save()

521 return {"deleted": True, "id": memory_id}

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

523

524 def _handle_update_memory(

525 self, memory_id: str, content: str = None, category: str = None, tags: list = None

526 ) -> dict:

527 e = self._entries.get(memory_id)

528 if not e:

529 return {"updated": False, "reason": "not found"}

530 if content is not None:

531 e["content"] = content

532 if category is not None:

533 e["category"] = category

534 if tags is not None:

535 e["tags"] = tags

536 e["updated_at"] = time.time()

537 self._save()

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

539

540

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

542

543

544class SearchServer:

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

546

547 NAME = "search"

548 VERSION = "1.0.0"

549

550 def get_tools(self) -> list[dict[str, Any]]:

551 return [

552 {

553 "name": "web_search",

554 "description": "Search the web",

555 "inputSchema": {

556 "type": "object",

557 "properties": {

558 "query": {"type": "string"},

559 "max_results": {"type": "integer", "default": 10},

560 },

561 "required": ["query"],

562 },

563 },

564 {

565 "name": "news_search",

566 "description": "Search for recent news",

567 "inputSchema": {

568 "type": "object",

569 "properties": {

570 "query": {"type": "string"},

571 "max_results": {"type": "integer", "default": 10},

572 },

573 "required": ["query"],

574 },

575 },

576 {

577 "name": "image_search",

578 "description": "Search for images",

579 "inputSchema": {

580 "type": "object",

581 "properties": {

582 "query": {"type": "string"},

583 "max_results": {"type": "integer", "default": 10},

584 },

585 "required": ["query"],

586 },

587 },

588 {

589 "name": "suggest",

590 "description": "Get search autocomplete suggestions",

591 "inputSchema": {

592 "type": "object",

593 "properties": {"query": {"type": "string"}},

594 "required": ["query"],

595 },

596 },

597 ]

598

599 def call_tool(self, tool_name: str, arguments: dict) -> Any:

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

601

602 def _handle_web_search(self, query: str, max_results: int = 10) -> list[dict]:

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

604 q = urllib.parse.quote_plus(query)

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

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

607 try:

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

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

610 except Exception as e:

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

612

613 results = []

614 # Parse DuckDuckGo HTML results

615 for m in re.finditer(

616 r'<a[^>]*class="result__a"[^>]*href="([^"]*)"[^>]*>(.*?)</a>', html, re.DOTALL

617 ):

618 if len(results) >= max_results:

619 break

620 link = m.group(1)

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

622 # Find snippet

623 snippet = ""

624 sn_match = re.search(

625 r'<a[^>]*class="result__snippet"[^>]*>(.*?)</a>',

626 html[m.end() : m.end() + 500],

627 re.DOTALL,

628 )

629 if sn_match:

630 snippet = re.sub(r"<[^>]+>", "", sn_match.group(1)).strip()

631 results.append(

632 {"title": title, "url": link, "snippet": snippet, "source": "duckduckgo"}

633 )

634 return results

635

636 def _handle_news_search(self, query: str, max_results: int = 10) -> list[dict]:

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

638 return self._handle_web_search(q, max_results)

639

640 def _handle_image_search(self, query: str, max_results: int = 10) -> list[dict]:

641 q = urllib.parse.quote_plus(query)

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

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

644 try:

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

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

647 except Exception as e:

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

649

650 # Extract vqd token for image search

651 vqd = ""

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

653 if vqd_m:

654 vqd = vqd_m.group(1)

655

656 if vqd:

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

658 try:

659 with urllib.request.urlopen(

660 urllib.request.Request(img_url, headers={"User-Agent": "Mozilla/5.0"}),

661 timeout=15,

662 ) as r:

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

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

665 return [

666 {

667 "title": r.get("title", ""),

668 "url": r.get("url", ""),

669 "thumbnail": r.get("thumbnail", ""),

670 "source": "duckduckgo",

671 }

672 for r in results

673 ]

674 except:

675 pass

676 return []

677

678 def _handle_suggest(self, query: str) -> list[str]:

679 q = urllib.parse.quote_plus(query)

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

681 try:

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

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

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

685 except:

686 return []

687

688

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

690

691

692class GitServer:

693 """MCP-compatible git operations."""

694

695 NAME = "git"

696 VERSION = "1.0.0"

697

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

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

700 self._git: str | None = shutil.which("git")

701

702 def _run_git(self, *args) -> dict[str, Any]:

703 if not self._git:

704 return {"error": "git not installed"}

705 try:

706 r = subprocess.run(

707 [self._git, *args],

708 capture_output=True,

709 text=True,

710 cwd=str(self._repo_path),

711 timeout=30,

712 )

713 return {

714 "stdout": r.stdout.strip(),

715 "stderr": r.stderr.strip(),

716 "returncode": r.returncode,

717 }

718 except subprocess.TimeoutExpired:

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

720 except Exception as e:

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

722

723 def get_tools(self) -> list[dict[str, Any]]:

724 return [

725 {

726 "name": "git_status",

727 "description": "Show working tree status",

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

729 },

730 {

731 "name": "git_log",

732 "description": "Show commit history",

733 "inputSchema": {

734 "type": "object",

735 "properties": {

736 "max_count": {"type": "integer", "default": 10},

737 "oneline": {"type": "boolean", "default": True},

738 },

739 },

740 },

741 {

742 "name": "git_diff",

743 "description": "Show changes between commits/working tree",

744 "inputSchema": {

745 "type": "object",

746 "properties": {

747 "staged": {"type": "boolean", "default": False},

748 "commit": {"type": "string"},

749 },

750 },

751 },

752 {

753 "name": "git_branch",

754 "description": "List branches",

755 "inputSchema": {

756 "type": "object",

757 "properties": {"remote": {"type": "boolean", "default": False}},

758 },

759 },

760 ]

761

762 def call_tool(self, tool_name: str, arguments: dict) -> Any:

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

764

765 def _handle_git_status(self) -> dict:

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

767

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

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

770 if oneline:

771 args.append("--oneline")

772 return self._run_git(*args)

773

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

775 args = ["diff"]

776 if staged:

777 args.append("--staged")

778 if commit:

779 args.append(commit)

780 return self._run_git(*args)

781

782 def _handle_git_branch(self, remote: bool = False) -> dict:

783 args = ["branch"]

784 if remote:

785 args.append("-r")

786 return self._run_git(*args)

787

788

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

790

791

792class ShellServer:

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

794

795 NAME = "shell"

796 VERSION = "1.0.0"

797

798 SAFE_COMMANDS = {

799 "ls",

800 "cat",

801 "head",

802 "tail",

803 "wc",

804 "grep",

805 "find",

806 "du",

807 "df",

808 "echo",

809 "date",

810 "whoami",

811 "uname",

812 "pwd",

813 "which",

814 "env",

815 "ps",

816 "top",

817 "htop",

818 "tree",

819 "file",

820 "stat",

821 "md5sum",

822 "sha256sum",

823 "python3",

824 "python",

825 "pip",

826 "npm",

827 "node",

828 "curl",

829 "wget",

830 }

831

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

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

834 return base in self.SAFE_COMMANDS

835

836 def get_tools(self) -> list[dict[str, Any]]:

837 return [

838 {

839 "name": "run_command",

840 "description": "Execute a safe shell command",

841 "inputSchema": {

842 "type": "object",

843 "properties": {

844 "command": {"type": "string"},

845 "timeout": {"type": "integer", "default": 30},

846 },

847 "required": ["command"],

848 },

849 },

850 {

851 "name": "system_info",

852 "description": "Get system information (OS, CPU, memory)",

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

854 },

855 {

856 "name": "disk_usage",

857 "description": "Show disk usage for a path",

858 "inputSchema": {

859 "type": "object",

860 "properties": {"path": {"type": "string", "default": "."}},

861 },

862 },

863 ]

864

865 def call_tool(self, tool_name: str, arguments: dict) -> Any:

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

867

868 def _handle_run_command(self, command: str, timeout: int = 30) -> dict:

869 if not self._is_safe(command):

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

871 try:

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

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

874 except subprocess.TimeoutExpired:

875 return {"error": "Timeout"}

876 except Exception as e:

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

878

879 def _handle_system_info(self) -> dict:

880 import platform

881

882 return {

883 "os": platform.system(),

884 "release": platform.release(),

885 "version": platform.version(),

886 "machine": platform.machine(),

887 "processor": platform.processor(),

888 "python": platform.python_version(),

889 "hostname": platform.node(),

890 }

891

892 def _handle_disk_usage(self, path: str = ".") -> dict:

893 try:

894 usage = shutil.disk_usage(path)

895 return {

896 "path": path,

897 "total_gb": round(usage.total / 1024**3, 2),

898 "used_gb": round(usage.used / 1024**3, 2),

899 "free_gb": round(usage.free / 1024**3, 2),

900 }

901 except Exception as e:

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

903

904

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

906

907

908class CodeServer:

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

910

911 NAME = "code"

912 VERSION = "1.0.0"

913

914 def get_tools(self) -> list[dict[str, Any]]:

915 return [

916 {

917 "name": "run_python",

918 "description": "Execute Python code in sandbox",

919 "inputSchema": {

920 "type": "object",

921 "properties": {

922 "code": {"type": "string"},

923 "timeout": {"type": "integer", "default": 10},

924 },

925 "required": ["code"],

926 },

927 },

928 {

929 "name": "run_shell",

930 "description": "Execute a one-liner bash command",

931 "inputSchema": {

932 "type": "object",

933 "properties": {

934 "command": {"type": "string"},

935 "timeout": {"type": "integer", "default": 10},

936 },

937 "required": ["command"],

938 },

939 },

940 {

941 "name": "lint_code",

942 "description": "Basic code linting (syntax check)",

943 "inputSchema": {

944 "type": "object",

945 "properties": {

946 "code": {"type": "string"},

947 "language": {"type": "string", "default": "python"},

948 },

949 "required": ["code"],

950 },

951 },

952 ]

953

954 def call_tool(self, tool_name: str, arguments: dict) -> Any:

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

956

957 def _handle_run_python(self, code: str, timeout: int = 10) -> dict:

958 try:

959 # Restricted execution via compile + eval in limited namespace

960 restricted_globals = {

961 "__builtins__": {

962 "print": print,

963 "len": len,

964 "range": range,

965 "int": int,

966 "float": float,

967 "str": str,

968 "list": list,

969 "dict": dict,

970 "bool": bool,

971 "set": set,

972 "tuple": tuple,

973 "sum": sum,

974 "min": min,

975 "max": max,

976 "abs": abs,

977 "round": round,

978 "sorted": sorted,

979 "enumerate": enumerate,

980 "zip": zip,

981 "map": map,

982 "filter": filter,

983 "json": __import__("json"),

984 "math": __import__("math"),

985 "datetime": __import__("datetime"),

986 "re": __import__("re"),

987 "collections": __import__("collections"),

988 "itertools": __import__("itertools"),

989 }

990 }

991 import io

992 import sys

993

994 old_stdout = sys.stdout

995 sys.stdout = buffer = io.StringIO()

996 try:

997 compiled = compile(code, "<mcp_sandbox>", "exec")

998 exec(compiled, restricted_globals)

999 output = buffer.getvalue()

1000 finally:

1001 sys.stdout = old_stdout

1002 return {"output": output, "success": True}

1003 except Exception as e:

1004 return {"output": str(e), "success": False, "error": type(e).__name__}

1005

1006 def _handle_run_shell(self, command: str, timeout: int = 10) -> dict:

1007 try:

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

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

1010 except subprocess.TimeoutExpired:

1011 return {"error": "Timeout"}

1012 except Exception as e:

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

1014

1015 def _handle_lint_code(self, code: str, language: str = "python") -> dict:

1016 if language == "python":

1017 try:

1018 compile(code, "<lint>", "exec")

1019 return {"valid": True, "errors": []}

1020 except SyntaxError as e:

1021 return {

1022 "valid": False,

1023 "errors": [{"line": e.lineno, "offset": e.offset, "message": e.msg}],

1024 }

1025 return {"valid": None, "message": f"Linting not supported for {language}"}

1026

1027

1028# ── Text Server (4 tools) ────────────────────

1029

1030

1031class TextServer:

1032 """MCP-compatible text manipulation tools."""

1033

1034 NAME = "text"

1035 VERSION = "1.0.0"

1036

1037 def get_tools(self) -> list[dict[str, Any]]:

1038 return [

1039 {

1040 "name": "count_tokens",

1041 "description": "Estimate token count (OpenAI tiktoken-style approximate)",

1042 "inputSchema": {

1043 "type": "object",

1044 "properties": {"text": {"type": "string"}},

1045 "required": ["text"],

1046 },

1047 },

1048 {

1049 "name": "extract_regex",

1050 "description": "Extract patterns from text using regex",

1051 "inputSchema": {

1052 "type": "object",

1053 "properties": {

1054 "text": {"type": "string"},

1055 "pattern": {"type": "string"},

1056 "group": {"type": "integer", "default": 0},

1057 },

1058 "required": ["text", "pattern"],

1059 },

1060 },

1061 {

1062 "name": "summarize_text",

1063 "description": "Simple extractive text summarization",

1064 "inputSchema": {

1065 "type": "object",

1066 "properties": {

1067 "text": {"type": "string"},

1068 "max_sentences": {"type": "integer", "default": 5},

1069 },

1070 "required": ["text"],

1071 },

1072 },

1073 {

1074 "name": "format_json",

1075 "description": "Format/validate/prettify JSON",

1076 "inputSchema": {

1077 "type": "object",

1078 "properties": {

1079 "text": {"type": "string"},

1080 "indent": {"type": "integer", "default": 2},

1081 },

1082 "required": ["text"],

1083 },

1084 },

1085 ]

1086

1087 def call_tool(self, tool_name: str, arguments: dict) -> Any:

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

1089

1090 def _handle_count_tokens(self, text: str) -> dict:

1091 # Approximate: ~4 chars per token for English, ~1.5 for CJK

1092 words = len(re.findall(r"\w+", text))

1093 chars = len(text)

1094 return {"tokens_approx": max(1, words + chars // 4), "characters": chars, "words": words}

1095

1096 def _handle_extract_regex(self, text: str, pattern: str, group: int = 0) -> list[str]:

1097 try:

1098 return [m.group(group) if group else m.group(0) for m in re.finditer(pattern, text)]

1099 except re.error as e:

1100 return [f"Invalid regex: {e}"]

1101

1102 def _handle_summarize_text(self, text: str, max_sentences: int = 5) -> str:

1103 sentences = re.split(r"(?<=[.!?])\s+", text)

1104 if len(sentences) <= max_sentences:

1105 return text

1106 # Simple extractive: take first sentence + longest sentences

1107 first = sentences[0]

1108 rest = sorted(sentences[1:], key=len, reverse=True)[: max_sentences - 1]

1109 return ". ".join([first] + rest) + "."

1110

1111 def _handle_format_json(self, text: str, indent: int = 2) -> dict:

1112 try:

1113 data = json.loads(text)

1114 formatted = json.dumps(data, indent=indent, ensure_ascii=False)

1115 return {

1116 "valid": True,

1117 "formatted": formatted,

1118 "keys": list(data.keys()) if isinstance(data, dict) else None,

1119 }

1120 except json.JSONDecodeError as e:

1121 return {"valid": False, "error": str(e)}

1122

1123

1124# ── Built-in Server Registry ─────────────────

1125

1126

1127class BuiltinMCPRegistry:

1128 """Registry of all built-in MCP servers. Single interface for tool discovery/calling."""

1129

1130 def __init__(self):

1131 self._servers: dict[str, Any] = {}

1132

1133 def register_server(self, server: Any) -> None:

1134 self._servers[server.NAME] = server

1135

1136 def list_all_tools(self) -> list[dict[str, Any]]:

1137 tools = []

1138 for srv_name, server in self._servers.items():

1139 for tool in server.get_tools():

1140 tools.append(

1141 {

1142 "server": srv_name,

1143 "name": f"mcp__{srv_name}__{tool['name']}",

1144 "description": tool.get("description", ""),

1145 "inputSchema": tool.get("inputSchema", {}),

1146 }

1147 )

1148 return tools

1149

1150 def get_tool_schemas(self, format: str = "openai") -> list[dict[str, Any]]:

1151 schemas = []

1152 for srv_name, server in self._servers.items():

1153 for tool in server.get_tools():

1154 schemas.append(

1155 {

1156 "type": "function",

1157 "function": {

1158 "name": f"mcp__{srv_name}__{tool['name']}",

1159 "description": tool.get("description", ""),

1160 "parameters": tool.get("inputSchema", {}),

1161 },

1162 }

1163 )

1164 return schemas

1165

1166 def call_tool(self, server_name: str, tool_name: str, arguments: dict[str, Any]) -> Any:

1167 server = self._servers.get(server_name)

1168 if not server:

1169 raise ValueError(f"Server '{server_name}' not found")

1170 return server.call_tool(tool_name, arguments)

1171

1172 def call_tool_by_full_name(self, full_name: str, arguments: dict[str, Any]) -> Any:

1173 parts = full_name.split("__", 2)

1174 if len(parts) != 3 or parts[0] != "mcp":

1175 raise ValueError(f"Invalid tool name: {full_name}")

1176 return self.call_tool(parts[1], parts[2], arguments)

1177

1178 @property

1179 def server_names(self) -> list[str]:

1180 return list(self._servers.keys())

1181

1182 @property

1183 def tool_count(self) -> int:

1184 return sum(len(s.get_tools()) for s in self._servers.values())

1185

1186

1187def create_default_registry(

1188 allowed_paths: list[str] | None = None,

1189 memory_path: str | None = None,

1190 repo_path: str = "",

1191) -> BuiltinMCPRegistry:

1192 """Create a BuiltinMCPRegistry with all 8 servers registered."""

1193 if allowed_paths is None:

1194 allowed_paths = [os.getcwd(), str(Path.home())]

1195 reg = BuiltinMCPRegistry()

1196 reg.register_server(FilesystemServer(allowed_paths=allowed_paths))

1197 reg.register_server(WebFetchServer())

1198 reg.register_server(MemoryServer(storage_path=memory_path or ""))

1199 reg.register_server(SearchServer())

1200 reg.register_server(GitServer(repo_path=repo_path))

1201 reg.register_server(ShellServer())

1202 reg.register_server(CodeServer())

1203 reg.register_server(TextServer())

1204 return reg

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

459 statements