#!/usr/bin/env python3
"""
MCP Smoke Test
Validates that the RLM MCP server starts, speaks JSON-RPC, and exposes the expected tools.
"""
import json
import os
import subprocess
import sys
import time
from typing import Dict, Any, Optional
def run_smoketest():
# 1. Launch Server
# We use sys.executable to ensure we use the same python environment (e.g. uv venv)
server_script = os.path.join(os.getcwd(), "rlm_mcp_server", "server.py")
if not os.path.exists(server_script):
print(f"Error: Server script not found at {server_script}")
sys.exit(1)
print(f"Starting server: {sys.executable} {server_script}")
process = subprocess.Popen(
[sys.executable, "-u", server_script], # -u for unbuffered binary stdout
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=sys.stderr,
text=True,
bufsize=0
)
def send_request(method: str, params: Optional[Dict] = None, msg_id: int = 1):
req = {
"jsonrpc": "2.0",
"method": method,
"id": msg_id
}
if params is not None:
req["params"] = params
json_req = json.dumps(req)
# print(f"> {json_req}", file=sys.stderr)
process.stdin.write(json_req + "\n")
process.stdin.flush()
def send_notification(method: str, params: Optional[Dict] = None):
req = {
"jsonrpc": "2.0",
"method": method
}
if params is not None:
req["params"] = params
json_req = json.dumps(req)
process.stdin.write(json_req + "\n")
process.stdin.flush()
def read_response(timeout=5) -> Dict[str, Any]:
start_time = time.time()
while time.time() - start_time < timeout:
line = process.stdout.readline()
if line:
# print(f"< {line.strip()}", file=sys.stderr)
return json.loads(line)
time.sleep(0.1)
raise TimeoutError("Timed out waiting for response")
try:
# 2. Handshake
print("Sending initialize...")
send_request("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "smoketest", "version": "0.1.0"}
}, 1)
resp = read_response()
if "error" in resp:
print(f"Initialize failed: {resp['error']}")
sys.exit(1)
print("Initialize success. Protocol version:", resp["result"]["protocolVersion"])
send_notification("notifications/initialized")
# 3. List Tools
print("Listing tools...")
send_request("tools/list", {}, 2)
resp = read_response()
if "error" in resp:
print(f"tools/list failed: {resp['error']}")
sys.exit(1)
tools = resp["result"]["tools"]
tool_names = [t["name"] for t in tools]
print(f"Tools found: {tool_names}")
if "solve" not in tool_names:
print("Error: 'solve' tool missing!")
sys.exit(1)
# 4. Call Tool (Dry Run)
print("Calling tool 'solve' (dry run)...")
# Use a dummy call that should fail fast or return a simple result without needing real LLM keys
# The spec says: "tiny context and max_iterations=1"
send_request("tools/call", {
"name": "solve",
"arguments": {
"query": "Hello world",
"environment": "local", # Try to avoid docker requirement for smoke test if possible
"max_iterations": 1,
"text": "Context"
}
}, 3)
# Depending on environment, this might take a moment or fail if models aren't present.
# We mainly want to ensure the server doesn't crash and returns *something* valid JSON-RPC.
resp = read_response(timeout=30)
if "error" in resp:
# An application error is acceptable (e.g. "Model not found"),
# as long as the server handled the request.
print(f"Tool call returned error (expected in some envs): {resp['error']['message']}")
else:
print("Tool call successful.")
content = resp["result"].get("content", [])
for block in content:
if block["type"] == "text":
print(f"Response text preview: {block['text'][:100]}...")
print("\nSmoke test PASSED.")
except Exception as e:
print(f"\nSmoke test FAILED: {e}")
sys.exit(1)
finally:
process.terminate()
try:
process.wait(timeout=2)
except subprocess.TimeoutExpired:
process.kill()
if __name__ == "__main__":
run_smoketest()
