#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = ["mcp>=2.3,<3"]
# ///
"""Check an MCP server over Streamable HTTP: connect, list its tools, maybe call one.

From https://exe.dev/deploy/mcp-server. Usage, with uv installed:

  uv run check.py URL [-H 'Name: value']... [--call TOOL [JSON_ARGS]]
  ./check.py URL ...  (the same, once it's executable)

Exits non-zero if anything fails.
"""

import argparse
import asyncio
import json
import sys

import httpx2
from mcp.client import Client
from mcp.client.streamable_http import streamable_http_client


async def main() -> None:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument("url")
    p.add_argument("-H", "--header", action="append", default=[], help="extra request header")
    p.add_argument("--call", nargs="+", metavar=("TOOL", "JSON_ARGS"), help="call one tool")
    a = p.parse_args()

    headers = dict(h.split(":", 1) for h in a.header)
    headers = {k.strip(): v.strip() for k, v in headers.items()}
    http = httpx2.AsyncClient(headers=headers, timeout=httpx2.Timeout(30, read=300))
    async with Client(streamable_http_client(a.url, http_client=http)) as c:
        info = c.server_info
        print(f"connected: {info.name if info else '?'} (protocol {c.protocol_version})")
        tools = (await c.list_tools()).tools
        print(f"{len(tools)} tools: {', '.join(t.name for t in tools)}")
        if a.call:
            args = json.loads(a.call[1]) if len(a.call) > 1 else {}
            r = await c.call_tool(a.call[0], args)
            for block in r.content:
                print(getattr(block, "text", block))
            if r.is_error:
                sys.exit(f"{a.call[0]} returned an error")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:  # one line, not a traceback, for people and agents alike
        msg = f"{type(e).__name__}: {e}"
        if isinstance(e, BaseExceptionGroup):
            msg = "; ".join(f"{type(x).__name__}: {x}" for x in e.exceptions)
        sys.exit(f"check failed: {msg}")