Coverage for little_loops / mcp_call.py: 0%

1"""mcp-call — thin CLI wrapper for direct MCP tool invocation. 

2 

3Usage: 

4 mcp-call server/tool-name '{"param": "value"}' 

5 

6Reads .mcp.json from the current working directory, spawns the MCP server 

7subprocess, performs the JSON-RPC initialize handshake, calls tools/call, 

8and writes the MCP response envelope to stdout. 

9 

10Exit codes: 

11 0 → success (isError: false) 

12 1 → tool_error (isError: true) 

13 124 → timeout (transport-level) 

14 127 → not_found (server or tool missing from .mcp.json) 

15 2 → usage/config error 

16""" 

17 

18from __future__ import annotations 

19 

20import json 

21import subprocess 

22import sys 

23import threading 

24import time 

25from pathlib import Path 

26from typing import Any 

27 

28MCP_PROTOCOL_VERSION = "2024-11-05" 

29_DEFAULT_TIMEOUT = 30 # seconds 

30 

31 

32def _load_mcp_config(cwd: Path) -> dict[str, Any]: 

33 """Load .mcp.json from the current working directory. 

34 

35 Args: 

36 cwd: Directory to search for .mcp.json 

37 

38 Returns: 

39 Parsed .mcp.json content 

40 

41 Raises: 

42 FileNotFoundError: If .mcp.json not found 

43 json.JSONDecodeError: If .mcp.json is invalid JSON 

44 """ 

45 mcp_path = cwd / ".mcp.json" 

46 if not mcp_path.exists(): 

47 raise FileNotFoundError(f".mcp.json not found in {cwd}") 

48 result: dict[str, Any] = json.loads(mcp_path.read_text()) 

49 return result 

50 

51 

52def _find_server_config(mcp_config: dict[str, Any], server_name: str) -> dict[str, Any] | None: 

53 """Find server configuration by name in .mcp.json. 

54 

55 Args: 

56 mcp_config: Parsed .mcp.json content 

57 server_name: Server name to find 

58 

59 Returns: 

60 Server config dict, or None if not found 

61 """ 

62 servers: dict[str, Any] = mcp_config.get("mcpServers", {}) 

63 found: dict[str, Any] | None = servers.get(server_name) 

64 return found 

65 

66 

67def _send_jsonrpc( 

68 proc: subprocess.Popen[str], 

69 request: dict[str, Any], 

70 request_id: int | None, 

71 timeout: float, 

72) -> dict[str, Any] | None: 

73 """Send a JSON-RPC request and optionally wait for a response. 

74 

75 Args: 

76 proc: Running MCP server process 

77 request: JSON-RPC request dict 

78 request_id: Expected response id (None for notifications) 

79 timeout: Seconds to wait for response 

80 

81 Returns: 

82 Parsed response dict, or None for notifications 

83 """ 

84 assert proc.stdin is not None 

85 line = json.dumps(request) + "\n" 

86 proc.stdin.write(line) 

87 proc.stdin.flush() 

88 

89 if request_id is None: 

90 # Notification — no response expected 

91 return None 

92 

93 # Read response lines until we find the matching id or timeout 

94 assert proc.stdout is not None 

95 deadline = time.monotonic() + timeout 

96 while time.monotonic() < deadline: 

97 proc.stdout.readable() 

98 try: 

99 response_line = proc.stdout.readline() 

100 except (OSError, ValueError): 

101 break 

102 if not response_line: 

103 break 

104 response_line = response_line.strip() 

105 if not response_line: 

106 continue 

107 try: 

108 response: dict[str, Any] = json.loads(response_line) 

109 if response.get("id") == request_id: 

110 return response 

111 except json.JSONDecodeError: 

112 continue 

113 

114 return None 

115 

116 

117def call_mcp_tool( 

118 server_name: str, 

119 tool_name: str, 

120 params: dict[str, Any], 

121 timeout: int = _DEFAULT_TIMEOUT, 

122 cwd: Path | None = None, 

123) -> tuple[dict[str, Any], int]: 

124 """Call an MCP tool via JSON-RPC and return the response envelope + exit code. 

125 

126 Args: 

127 server_name: MCP server name (from .mcp.json mcpServers key) 

128 tool_name: Tool name to call 

129 params: Tool arguments 

130 timeout: Timeout in seconds 

131 cwd: Directory containing .mcp.json (defaults to Path.cwd()) 

132 

133 Returns: 

134 Tuple of (MCP response envelope dict, exit code) 

135 Exit codes: 

136 0 → success 

137 1 → tool_error 

138 124 → timeout 

139 127 → not_found 

140 2 → config/usage error 

141 """ 

142 if cwd is None: 

143 cwd = Path.cwd() 

144 

145 # Load .mcp.json 

146 try: 

147 mcp_config = _load_mcp_config(cwd) 

148 except FileNotFoundError as e: 

149 return {"isError": True, "content": [{"type": "text", "text": str(e)}]}, 127 

150 except json.JSONDecodeError as e: 

151 return { 

152 "isError": True, 

153 "content": [{"type": "text", "text": f"Invalid .mcp.json: {e}"}], 

154 }, 2 

155 

156 # Find server config 

157 server_config = _find_server_config(mcp_config, server_name) 

158 if server_config is None: 

159 return { 

160 "isError": True, 

161 "content": [{"type": "text", "text": f"Server '{server_name}' not found in .mcp.json"}], 

162 }, 127 

163 

164 # Build server command 

165 command = server_config.get("command") 

166 if not command: 

167 return { 

168 "isError": True, 

169 "content": [ 

170 {"type": "text", "text": f"Server '{server_name}' has no 'command' in .mcp.json"} 

171 ], 

172 }, 2 

173 

174 args = server_config.get("args", []) 

175 env_overrides = server_config.get("env", {}) 

176 

177 import os 

178 

179 env = os.environ.copy() 

180 env.update({k: str(v) for k, v in env_overrides.items()}) 

181 

182 cmd = [command] + args 

183 

184 # Spawn server 

185 try: 

186 proc = subprocess.Popen( 

187 cmd, 

188 stdin=subprocess.PIPE, 

189 stdout=subprocess.PIPE, 

190 stderr=subprocess.PIPE, 

191 text=True, 

192 env=env, 

193 ) 

194 except FileNotFoundError: 

195 return { 

196 "isError": True, 

197 "content": [{"type": "text", "text": f"Server command not found: {command}"}], 

198 }, 127 

199 

200 stderr_chunks: list[str] = [] 

201 

202 def _drain_stderr() -> None: 

203 assert proc.stderr is not None 

204 for line in proc.stderr: 

205 stderr_chunks.append(line) 

206 

207 stderr_thread = threading.Thread(target=_drain_stderr, daemon=True) 

208 stderr_thread.start() 

209 

210 try: 

211 # Step 1: Send initialize request 

212 init_response = _send_jsonrpc( 

213 proc, 

214 { 

215 "jsonrpc": "2.0", 

216 "id": 1, 

217 "method": "initialize", 

218 "params": { 

219 "protocolVersion": MCP_PROTOCOL_VERSION, 

220 "capabilities": {}, 

221 "clientInfo": {"name": "mcp-call", "version": "1.0"}, 

222 }, 

223 }, 

224 request_id=1, 

225 timeout=timeout, 

226 ) 

227 

228 if init_response is None: 

229 proc.kill() 

230 return { 

231 "isError": True, 

232 "content": [ 

233 {"type": "text", "text": "No response to initialize request (timeout)"} 

234 ], 

235 }, 124 

236 

237 if "error" in init_response: 

238 proc.kill() 

239 return { 

240 "isError": True, 

241 "content": [ 

242 {"type": "text", "text": f"Initialize failed: {init_response['error']}"} 

243 ], 

244 }, 1 

245 

246 # Step 2: Send initialized notification 

247 _send_jsonrpc( 

248 proc, 

249 {"jsonrpc": "2.0", "method": "notifications/initialized"}, 

250 request_id=None, 

251 timeout=0, 

252 ) 

253 

254 # Step 3: Call tools/call 

255 call_response = _send_jsonrpc( 

256 proc, 

257 { 

258 "jsonrpc": "2.0", 

259 "id": 2, 

260 "method": "tools/call", 

261 "params": {"name": tool_name, "arguments": params}, 

262 }, 

263 request_id=2, 

264 timeout=timeout, 

265 ) 

266 

267 if call_response is None: 

268 proc.kill() 

269 return { 

270 "isError": True, 

271 "content": [ 

272 {"type": "text", "text": "No response to tools/call request (timeout)"} 

273 ], 

274 }, 124 

275 

276 if "error" in call_response: 

277 err = call_response["error"] 

278 # JSON-RPC method not found → tool not found 

279 if isinstance(err, dict) and err.get("code") == -32601: 

280 return { 

281 "isError": True, 

282 "content": [ 

283 { 

284 "type": "text", 

285 "text": f"Tool '{tool_name}' not found on server '{server_name}'", 

286 } 

287 ], 

288 }, 127 

289 return { 

290 "isError": True, 

291 "content": [{"type": "text", "text": f"tools/call error: {err}"}], 

292 }, 1 

293 

294 # Unwrap result from JSON-RPC envelope 

295 result = call_response.get("result", {}) 

296 is_error = result.get("isError", False) 

297 exit_code = 1 if is_error else 0 

298 return result, exit_code 

299 

300 except subprocess.TimeoutExpired: 

301 proc.kill() 

302 return { 

303 "isError": True, 

304 "content": [{"type": "text", "text": "MCP tool call timed out"}], 

305 }, 124 

306 finally: 

307 try: 

308 proc.terminate() 

309 proc.wait(timeout=5) 

310 except Exception: 

311 proc.kill() 

312 stderr_thread.join(timeout=5) 

313 

314 

315def main() -> None: 

316 """Entry point for mcp-call CLI.""" 

317 if len(sys.argv) < 3: 

318 print( 

319 'Usage: mcp-call server/tool-name \'{"param": "value"}\'', 

320 file=sys.stderr, 

321 ) 

322 sys.exit(2) 

323 

324 spec = sys.argv[1] 

325 params_json = sys.argv[2] 

326 

327 # Parse server/tool-name 

328 if "/" not in spec: 

329 print( 

330 f"Error: spec must be 'server/tool-name', got: {spec!r}", 

331 file=sys.stderr, 

332 ) 

333 sys.exit(2) 

334 

335 server_name, tool_name = spec.split("/", 1) 

336 

337 # Parse params JSON 

338 try: 

339 params = json.loads(params_json) 

340 except json.JSONDecodeError as e: 

341 print(f"Error: invalid params JSON: {e}", file=sys.stderr) 

342 sys.exit(2) 

343 

344 if not isinstance(params, dict): 

345 print("Error: params must be a JSON object", file=sys.stderr) 

346 sys.exit(2) 

347 

348 envelope, exit_code = call_mcp_tool( 

349 server_name=server_name, 

350 tool_name=tool_name, 

351 params=params, 

352 ) 

353 

354 print(json.dumps(envelope)) 

355 sys.exit(exit_code) 

356 

357 

358if __name__ == "__main__": 

359 main()