MCP Inspector: Connect, Test and Debug an MCP Server

#!/usr/bin/env python3
"""Smoke-test an MCP server through the MCP Inspector CLI (stdlib only).

  inspector_smoke.py [options] URL
  inspector_smoke.py [options] -- COMMAND [ARG...]

Checks, in order: inspector (version), connect (initialize), era (which protocol version was
negotiated), tools-list (tools/list --strict), tools-call (with --call), stdout-clean (with
--check-stdout, stdio only).

Options:
  --inspector CMD    how to start the Inspector (default: npx -y @modelcontextprotocol/[email protected])
  --era ERA          legacy | auto | modern (default: modern)
  --transport T      http | sse, for a URL target (default: http)
  --timeout-ms N     the Inspector's --connect-timeout, in milliseconds (default: 15000)
  --call NAME        also call this tool once, with the --arg values
  --arg KEY=VALUE    argument for --call (repeatable); a value is sent as JSON when it parses
  --expect-error     the --call is expected to return isError: true
  --check-stdout     stdio only: drive the server by hand and require stdout to carry only JSON-RPC

Exit status: 0 no FAIL (WARN allowed), 1 at least one FAIL, 2 usage error or Inspector cannot run.
"""
import argparse
import json
import queue
import shlex
import subprocess
import sys
import textwrap
import threading

PROCESS_TIMEOUT_S = 90
RAW_TIMEOUT_S = 10
MODERN_VERSION = "2026-07-28"
results = []


def report(level, name, detail):
    """Record and print one check result."""
    results.append(level)
    print(f"{level} {name}: {detail}")


def parse_json(text):
    """Parse text as JSON; return None when it is not JSON."""
    try:
        return json.loads(text)
    except ValueError:
        return None


def run_cli(base, args):
    """Run the Inspector CLI once; return (exit code, stdout, stderr), exit code None on timeout."""
    try:
        done = subprocess.run(base + ["--cli"] + args, capture_output=True, text=True,
                              stdin=subprocess.DEVNULL, timeout=PROCESS_TIMEOUT_S)
    except subprocess.TimeoutExpired:
        return None, "", ""
    return done.returncode, done.stdout, done.stderr


def error_envelope(stderr):
    """The Inspector writes a one-line JSON error envelope last on stderr; return its dict or {}."""
    for line in reversed(stderr.strip().splitlines()):
        value = parse_json(line)
        if isinstance(value, dict):
            return value.get("error") or {}
    return {}


def server_stderr(stderr):
    """The last stderr line that is not the error envelope: what a stdio server wrote before it died."""
    lines = [line for line in stderr.strip().splitlines() if line.strip() and not line.startswith('{"error"')]
    return textwrap.shorten(lines[-1], 80, placeholder="...") if lines else ""


def explain(code, stderr):
    """One-line description of a failed run: exit code, error code, message and what to check."""
    if code is None:
        return f"no result within {PROCESS_TIMEOUT_S}s (the Inspector or the server hangs)"
    err = error_envelope(stderr)
    message = textwrap.shorten(str(err.get("message", "")), 150, placeholder="...")
    hint = ""
    if "timed out" in message:
        hint = "the server did not answer in time: raise --timeout-ms, or check that it speaks MCP on this transport"
    elif code == 4:
        hint = "is the server running and the URL or command right?"
    elif code == 3:
        hint = "the server wants credentials (HTTP 401 or 403)"
    elif err.get("status") == 404:
        hint = "the endpoint answered 404: wrong path?"
    elif "Unexpected content type" in message:
        hint = "the URL answers with something that is not MCP (login page, proxy?)"
    elif "Connection closed" in message:
        hint = "the server process exited during startup: run it by hand and read its stderr"
    elif "ENOENT" in message:
        hint = "the command was not found"
    elif "Error POSTing" in message:
        hint = "the server answered with an HTTP error status"
    elif "Version negotiation failed" in message:
        hint = "the server does not speak 2026-07-28; try --era legacy"
    text = f"exit {code} {err.get('code', '-')}: {message}"
    if server_stderr(stderr):
        text += f"; stderr: {server_stderr(stderr)!r}"
    return f"{text} -> {hint}" if hint else text


def coerce(value):
    """Mirror the Inspector's --tool-arg rule for the raw session: JSON when it parses, else a string."""
    try:
        return json.loads(value)
    except ValueError:
        return value


def first_text(stdout):
    """The first text block of a tools/call result, shortened, or an empty string."""
    content = (parse_json(stdout) or {}).get("content") or [{}]
    return textwrap.shorten(str(content[0].get("text", "")), 150, placeholder="...")


def check_connect(run):
    """initialize; return the result dict, or None when the connection failed."""
    code, out, err = run(["--method", "initialize"])
    info = parse_json(out) if code == 0 else None
    if not isinstance(info, dict):
        report("FAIL", "connect", explain(code, err))
        return None
    version = info.get("protocolVersion")
    name = (info.get("serverInfo") or {}).get("name")
    if not (isinstance(version, str) and name and isinstance(info.get("capabilities"), dict)):
        report("FAIL", "connect", "initialize result lacks serverInfo.name, protocolVersion or capabilities")
        return None
    report("PASS", "connect", f"server={name} protocolVersion={version}")
    return info


def check_era(version):
    """WARN when the 2026-07-28 behavior was not exercised."""
    if version == MODERN_VERSION:
        report("PASS", "era", f"negotiated {MODERN_VERSION}")
    else:
        report("WARN", "era", f"negotiated {version}; 2026-07-28 behavior not exercised (--era modern pins it)")


def describe(tool, finding):
    """A short label for one schema finding: tool, location and rule."""
    location = ".".join(part for part in (finding["schema"], finding["path"]) if part)
    return f"{tool} {location} ({finding['rule']})"


def listed_tools(body):
    """The tools array of a --format json tools/list output, or an empty list."""
    if not isinstance(body, dict):
        return []
    return (body.get("result") or {}).get("tools") or []


def check_tools_list(run, advertised):
    """tools/list with --strict; exit 6 means an error-severity schema portability finding.

    The Inspector does not send tools/list when the server did not advertise capabilities.tools: it prints an
    empty list and exits 0. An empty list from a server that did advertise it is repeated once before it is reported.
    """
    options = ["--method", "tools/list", "--strict", "--format", "json"]
    code, out, err = run(options)
    body, again = parse_json(out), False
    if code == 0 and advertised and not listed_tools(body):
        code, out, err = run(options)
        body, again = parse_json(out), True
    if code not in (0, 6) or not isinstance(body, dict):
        report("FAIL", "tools-list", explain(code, err))
        return
    tools = listed_tools(body)
    where = [(tool["toolName"], f) for tool in body.get("schemaFindings", []) for f in tool["findings"]]
    errors = [describe(n, f) for n, f in where if f["severity"] == "error"]
    warnings = [describe(n, f) for n, f in where if f["severity"] != "error"]
    if code == 6 or errors:
        report("FAIL", "tools-list", f"exit {code}, {len(errors)} portability error(s): " + "; ".join(errors))
    elif not advertised:
        report("WARN", "tools-list", "capabilities.tools is not advertised, so the Inspector lists no tools without asking")
    elif not tools:
        report("WARN", "tools-list", "the server lists no tools (asked twice)")
    elif warnings:
        report("WARN", "tools-list", f"{len(tools)} tools, {len(warnings)} portability warning(s): " + "; ".join(warnings))
    else:
        names = ", ".join(sorted(t.get("name", "?") for t in tools))
        report("PASS", "tools-list", f"{len(tools)} tools: {names}" + (" (second attempt)" if again else ""))


def check_tools_call(run, name, pairs, expect_error):
    """tools/call once; exit 5 is isError (tool_is_error) or an unknown tool (tool_not_found)."""
    options = ["--method", "tools/call", "--tool-name", name]
    for pair in pairs:
        options += ["--tool-arg", pair]
    code, out, err = run(options)
    kind = error_envelope(err).get("code")
    if code == 0 and not expect_error:
        report("PASS", "tools-call", f"{name} returned isError false")
    elif code == 0:
        report("FAIL", "tools-call", f"{name} succeeded but isError true was expected")
    elif code == 5 and kind == "tool_is_error" and expect_error:
        report("PASS", "tools-call", f"{name} returned isError true (exit 5 tool_is_error), as expected")
    elif code == 5 and kind == "tool_is_error":
        report("FAIL", "tools-call", f"{name} returned isError true (exit 5 tool_is_error): {first_text(out)}")
    else:
        report("FAIL", "tools-call", explain(code, err))


def is_jsonrpc(line):
    """True when the line is a JSON object with jsonrpc 2.0."""
    parsed = parse_json(line)
    return isinstance(parsed, dict) and parsed.get("jsonrpc") == "2.0"


def pump(stream, out):
    """Copy a stream into a queue, line by line, and end with None."""
    for line in stream:
        out.put(line.rstrip("\n"))
    out.put(None)


def check_stdout_clean(command, call, pairs):
    """Drive a stdio server by hand; every stdout line must be a JSON-RPC message."""
    messages = [
        {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
            "protocolVersion": "2025-11-25", "capabilities": {},
            "clientInfo": {"name": "inspector-smoke", "version": "1"}}},
        {"jsonrpc": "2.0", "method": "notifications/initialized"},
        {"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
    ]
    if call:
        arguments = {k: coerce(v) for k, v in (pair.split("=", 1) for pair in pairs)}
        messages.append({"jsonrpc": "2.0", "id": 3, "method": "tools/call",
                         "params": {"name": call, "arguments": arguments}})
    try:
        proc = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                stderr=subprocess.DEVNULL, text=True)
    except OSError as exc:
        report("FAIL", "stdout-clean", f"cannot start the command: {exc}")
        return
    lines, seen, answered = queue.Queue(), [], set()
    threading.Thread(target=pump, args=(proc.stdout, lines), daemon=True).start()
    why = ""
    try:
        for message in messages:
            proc.stdin.write(json.dumps(message) + "\n")
            proc.stdin.flush()
            while message.get("id") is not None and message["id"] not in answered:
                line = lines.get(timeout=RAW_TIMEOUT_S)
                if line is None:
                    raise EOFError
                if line.strip():
                    seen.append(line)
                    parsed = parse_json(line)
                    if isinstance(parsed, dict) and "id" in parsed and "method" not in parsed:
                        answered.add(parsed["id"])
    except queue.Empty:
        why = f"no response within {RAW_TIMEOUT_S}s"
    except (EOFError, BrokenPipeError):
        why = "the server closed stdout or its stdin before answering"
    finally:
        proc.kill()
        proc.wait()
    last_id = messages[-1]["id"]
    bad = [(i, line) for i, line in enumerate(seen, 1) if not is_jsonrpc(line)]
    if last_id not in answered:
        report("FAIL", "stdout-clean", f"request {last_id} not answered: {why}")
    elif bad:
        i, line = bad[0]
        report("FAIL", "stdout-clean", f"{len(bad)} of {len(seen)} stdout lines are not JSON-RPC; first is line {i}: {line[:60]!r}")
    else:
        report("PASS", "stdout-clean", f"{len(seen)} stdout lines, all JSON-RPC")


def main(argv):
    parser = argparse.ArgumentParser(usage=__doc__, add_help=False)
    parser.add_argument("--inspector", default="npx -y @modelcontextprotocol/[email protected]")
    parser.add_argument("--era", choices=["legacy", "auto", "modern"], default="modern")
    parser.add_argument("--transport", choices=["http", "sse"], default="http")
    parser.add_argument("--timeout-ms", type=int, default=15000)
    parser.add_argument("--call")
    parser.add_argument("--arg", action="append", default=[])
    parser.add_argument("--expect-error", action="store_true")
    parser.add_argument("--check-stdout", action="store_true")
    parser.add_argument("target", nargs="+")
    try:
        opts = parser.parse_args(argv)
    except SystemExit:
        return 2
    is_url = len(opts.target) == 1 and opts.target[0].startswith(("http://", "https://"))
    if (opts.check_stdout and is_url) or (opts.expect_error and not opts.call) \
            or any("=" not in pair for pair in opts.arg) or opts.timeout_ms < 0:
        sys.stderr.write(__doc__)
        return 2
    base = shlex.split(opts.inspector)
    try:
        version = subprocess.run(base + ["--cli", "-V"], capture_output=True, text=True,
                                 stdin=subprocess.DEVNULL, timeout=PROCESS_TIMEOUT_S)
    except (OSError, subprocess.TimeoutExpired) as exc:
        sys.stderr.write(f"cannot run the Inspector ({opts.inspector!r}): {exc}\n")
        return 2
    if version.returncode != 0:
        sys.stderr.write(f"the Inspector exited {version.returncode} on -V: {version.stderr.strip()[:200]}\n")
        return 2
    report("PASS", "inspector", version.stdout.strip())

    options = ["--protocol-era", opts.era, "--connect-timeout", str(opts.timeout_ms)]

    def run(method_options):
        """One CLI run against the target; the documented form puts the target before "--" and options after."""
        if is_url:
            return run_cli(base, ["--server-url", opts.target[0], "--transport", opts.transport] + method_options + options)
        return run_cli(base, opts.target + ["--"] + method_options + options)

    info = check_connect(run)
    if info is not None:
        check_era(info["protocolVersion"])
        check_tools_list(run, "tools" in info["capabilities"])
        if opts.call:
            check_tools_call(run, opts.call, opts.arg, opts.expect_error)
    if opts.check_stdout:
        check_stdout_clean(opts.target, opts.call, opts.arg)
    print(f"{results.count('PASS')} pass, {results.count('WARN')} warn, {results.count('FAIL')} fail")
    return 1 if "FAIL" in results else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

"""Reference server on the official Python SDK (mcp==2.3.0) for the Inspector checks.

  sdk_server.py http PORT           Streamable HTTP at http://127.0.0.1:PORT/mcp, no auth
  sdk_server.py stdio [MODE] [--name NAME]
                                    stdio; MODE is one of the values below; NAME is the serverInfo name
                                    (default inspector-ref), a flag that must reach the server itself

MODE (stdio only):
  ok          well-behaved server
  banner      fault: prints a line to stdout (flushed) before the protocol starts
  noisy-tool  the add tool prints to stdout (flushed) while it runs; SDK 2.3.0 reroutes that to stderr
  crash       fault: writes to stderr and exits with status 3 before serving
  silent      fault: never answers (sleeps)
"""
import sys
import time
from typing import Annotated

USAGE = __doc__
MODES = ("ok", "banner", "noisy-tool", "crash", "silent")


def parse_args(argv):
    """Return (transport, port, mode, name) or exit with status 2 and the usage text."""
    name = "inspector-ref"
    if len(argv) >= 2 and argv[-2] == "--name":
        argv, name = argv[:-2], argv[-1]
    if len(argv) == 2 and argv[0] == "http" and argv[1].isdigit():
        return "http", int(argv[1]), "ok", name
    if argv and argv[0] == "stdio" and len(argv) <= 2:
        mode = argv[1] if len(argv) == 2 else "ok"
        if mode in MODES:
            return "stdio", 0, mode, name
    sys.stderr.write(USAGE)
    sys.exit(2)


transport, port, mode, server_name = parse_args(sys.argv[1:])

if mode == "crash":
    sys.stderr.write("fatal: cannot open database\n")
    sys.exit(3)
if mode == "silent":
    time.sleep(600)
if mode == "banner":
    print("Server starting...", flush=True)

from mcp.server.mcpserver import MCPServer  # noqa: E402  (after the fault modes on purpose)
from pydantic import Field  # noqa: E402

server = MCPServer(server_name)


@server.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    if mode == "noisy-tool":
        print("debug: adding", a, b, flush=True)
    return a + b


@server.tool()
def get_weather(location: Annotated[str, Field(min_length=2)]) -> str:
    """Return fixed weather for a location."""
    return f"Partly cloudy in {location}"


@server.tool()
def explode() -> str:
    """Always raise."""
    raise RuntimeError("boom")


if __name__ == "__main__":
    if transport == "http":
        server.run("streamable-http", port=port)
    else:
        server.run("stdio")