|
| 1 | +"""Command line interface for MCPForge.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import argparse |
| 6 | +import json |
| 7 | +import subprocess |
| 8 | +import sys |
| 9 | +from pathlib import Path |
| 10 | +from typing import Any |
| 11 | + |
| 12 | +from .templates import GITIGNORE, STARTER_README, STARTER_SERVER |
| 13 | + |
| 14 | + |
| 15 | +class CheckError(RuntimeError): |
| 16 | + """Raised when server validation fails.""" |
| 17 | + |
| 18 | + |
| 19 | +def send_message(stream, message: dict[str, Any]) -> None: |
| 20 | + encoded = json.dumps(message).encode("utf-8") |
| 21 | + stream.write(f"Content-Length: {len(encoded)}\r\n\r\n".encode("utf-8")) |
| 22 | + stream.write(encoded) |
| 23 | + stream.flush() |
| 24 | + |
| 25 | + |
| 26 | +def read_message(stream) -> dict[str, Any]: |
| 27 | + content_length = None |
| 28 | + while True: |
| 29 | + line = stream.readline() |
| 30 | + if not line: |
| 31 | + raise CheckError("Server closed the stdio stream unexpectedly.") |
| 32 | + if line in (b"\r\n", b"\n"): |
| 33 | + break |
| 34 | + header = line.decode("utf-8").strip() |
| 35 | + if header.lower().startswith("content-length:"): |
| 36 | + content_length = int(header.split(":", 1)[1].strip()) |
| 37 | + if content_length is None: |
| 38 | + raise CheckError("Server response was missing a Content-Length header.") |
| 39 | + body = stream.read(content_length) |
| 40 | + return json.loads(body.decode("utf-8")) |
| 41 | + |
| 42 | + |
| 43 | +def create_project(path: Path, force: bool) -> None: |
| 44 | + if path.exists() and any(path.iterdir()) and not force: |
| 45 | + raise SystemExit( |
| 46 | + f"Refusing to initialize non-empty directory: {path}. Use --force to overwrite files." |
| 47 | + ) |
| 48 | + |
| 49 | + path.mkdir(parents=True, exist_ok=True) |
| 50 | + (path / "server.py").write_text(STARTER_SERVER, encoding="utf-8") |
| 51 | + (path / "README.md").write_text(STARTER_README, encoding="utf-8") |
| 52 | + (path / ".gitignore").write_text(GITIGNORE, encoding="utf-8") |
| 53 | + |
| 54 | + print(f"Created starter MCP server at {path}") |
| 55 | + print("Next steps:") |
| 56 | + print(f" cd {path}") |
| 57 | + print(" python server.py") |
| 58 | + print(" mcpforge check .") |
| 59 | + |
| 60 | + |
| 61 | +def run_check(path: Path) -> None: |
| 62 | + server_path = path / "server.py" if path.is_dir() else path |
| 63 | + if not server_path.exists(): |
| 64 | + raise SystemExit(f"Could not find server entry point at {server_path}") |
| 65 | + |
| 66 | + process = subprocess.Popen( |
| 67 | + [sys.executable, str(server_path)], |
| 68 | + cwd=str(server_path.parent), |
| 69 | + stdin=subprocess.PIPE, |
| 70 | + stdout=subprocess.PIPE, |
| 71 | + stderr=subprocess.PIPE, |
| 72 | + ) |
| 73 | + |
| 74 | + if process.stdin is None or process.stdout is None or process.stderr is None: |
| 75 | + raise SystemExit("Could not start validation process.") |
| 76 | + |
| 77 | + try: |
| 78 | + send_message( |
| 79 | + process.stdin, |
| 80 | + { |
| 81 | + "jsonrpc": "2.0", |
| 82 | + "id": 1, |
| 83 | + "method": "initialize", |
| 84 | + "params": { |
| 85 | + "protocolVersion": "2025-03-26", |
| 86 | + "capabilities": {}, |
| 87 | + "clientInfo": {"name": "mcpforge", "version": "0.1.0"}, |
| 88 | + }, |
| 89 | + }, |
| 90 | + ) |
| 91 | + initialize_response = read_message(process.stdout) |
| 92 | + result = initialize_response.get("result", {}) |
| 93 | + if "serverInfo" not in result or "capabilities" not in result: |
| 94 | + raise CheckError("Initialize response did not include serverInfo and capabilities.") |
| 95 | + |
| 96 | + send_message( |
| 97 | + process.stdin, |
| 98 | + {"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}, |
| 99 | + ) |
| 100 | + |
| 101 | + for request_id, method, expected_key in ( |
| 102 | + (2, "tools/list", "tools"), |
| 103 | + (3, "resources/list", "resources"), |
| 104 | + (4, "prompts/list", "prompts"), |
| 105 | + ): |
| 106 | + send_message( |
| 107 | + process.stdin, |
| 108 | + {"jsonrpc": "2.0", "id": request_id, "method": method, "params": {}}, |
| 109 | + ) |
| 110 | + response = read_message(process.stdout) |
| 111 | + payload = response.get("result", {}) |
| 112 | + if expected_key not in payload: |
| 113 | + raise CheckError(f"{method} did not return a '{expected_key}' payload.") |
| 114 | + |
| 115 | + server_info = result["serverInfo"] |
| 116 | + print("PASS: stdio server handshake succeeded.") |
| 117 | + print(f"Server: {server_info.get('name', 'unknown')} {server_info.get('version', '')}".strip()) |
| 118 | + print("Verified methods: initialize, tools/list, resources/list, prompts/list") |
| 119 | + except CheckError as exc: |
| 120 | + stderr_output = process.stderr.read().decode("utf-8").strip() |
| 121 | + print(f"FAIL: {exc}", file=sys.stderr) |
| 122 | + if stderr_output: |
| 123 | + print("\nServer stderr:\n" + stderr_output, file=sys.stderr) |
| 124 | + raise SystemExit(1) from exc |
| 125 | + finally: |
| 126 | + process.terminate() |
| 127 | + try: |
| 128 | + process.wait(timeout=2) |
| 129 | + except subprocess.TimeoutExpired: |
| 130 | + process.kill() |
| 131 | + |
| 132 | + |
| 133 | +def build_parser() -> argparse.ArgumentParser: |
| 134 | + parser = argparse.ArgumentParser(description="Scaffold and validate MCP server projects.") |
| 135 | + subparsers = parser.add_subparsers(dest="command", required=True) |
| 136 | + |
| 137 | + init_parser = subparsers.add_parser("init", help="Create a starter MCP server project.") |
| 138 | + init_parser.add_argument("path", help="Directory to create.") |
| 139 | + init_parser.add_argument("--force", action="store_true", help="Overwrite generated files in a non-empty directory.") |
| 140 | + |
| 141 | + check_parser = subparsers.add_parser("check", help="Run a local stdio smoke check against a server.") |
| 142 | + check_parser.add_argument("path", help="Project directory or server.py path.") |
| 143 | + |
| 144 | + return parser |
| 145 | + |
| 146 | + |
| 147 | +def main(argv: list[str] | None = None) -> int: |
| 148 | + parser = build_parser() |
| 149 | + args = parser.parse_args(argv) |
| 150 | + |
| 151 | + if args.command == "init": |
| 152 | + create_project(Path(args.path).resolve(), args.force) |
| 153 | + return 0 |
| 154 | + |
| 155 | + if args.command == "check": |
| 156 | + run_check(Path(args.path).resolve()) |
| 157 | + return 0 |
| 158 | + |
| 159 | + parser.error(f"Unknown command: {args.command}") |
| 160 | + return 1 |
| 161 | + |
| 162 | + |
| 163 | +if __name__ == "__main__": |
| 164 | + raise SystemExit(main()) |
0 commit comments