#!/usr/bin/env python3
"""Verify a Streamable HTTP MCP catalogue without invoking any tool."""

from __future__ import annotations

import argparse
import asyncio
import json

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client


def invalid_patterns(value, path: str = "schema") -> list[str]:
    issues = []
    if isinstance(value, dict):
        pattern = value.get("pattern")
        if isinstance(pattern, str) and not (
            pattern.startswith("^") and pattern.endswith("$")
        ):
            issues.append(f"{path}.pattern={pattern!r}")
        for key, item in value.items():
            issues.extend(invalid_patterns(item, f"{path}.{key}"))
    elif isinstance(value, list):
        for index, item in enumerate(value):
            issues.extend(invalid_patterns(item, f"{path}[{index}]"))
    return issues


async def verify(
    url: str,
    required: set[str],
    forbidden: set[str],
    max_tools: int,
    max_schema_chars: int,
    timeout: float,
) -> None:
    async with asyncio.timeout(timeout):
        async with streamablehttp_client(url) as (read_stream, write_stream, _):
            async with ClientSession(read_stream, write_stream) as session:
                await session.initialize()
                response = await session.list_tools()
    names = {tool.name for tool in response.tools}
    schemas = [tool.inputSchema for tool in response.tools]
    schema_chars = len(json.dumps(schemas, ensure_ascii=False, separators=(",", ":")))
    pattern_issues = [
        issue
        for index, schema in enumerate(schemas)
        for issue in invalid_patterns(schema, f"tools[{index}]")
    ]
    missing = required - names
    exposed = forbidden & names
    if missing or exposed or len(names) > max_tools or schema_chars > max_schema_chars or pattern_issues:
        raise SystemExit(
            f"MCP catalogue mismatch: missing={sorted(missing)}, "
            f"forbidden={sorted(exposed)}, count={len(names)}/{max_tools}, "
            f"schema_chars={schema_chars}/{max_schema_chars}, "
            f"invalid_patterns={pattern_issues}, actual={sorted(names)}"
        )
    print(
        f"MCP_CATALOG_OK count={len(names)} schema_chars={schema_chars} tools="
        + ",".join(sorted(names))
    )


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("url")
    parser.add_argument("--require", action="append", default=[])
    parser.add_argument("--forbid", action="append", default=[])
    parser.add_argument("--max-tools", type=int, default=64)
    parser.add_argument("--max-schema-chars", type=int, default=100_000)
    parser.add_argument("--timeout", type=float, default=10.0)
    args = parser.parse_args()
    asyncio.run(
        verify(
            args.url,
            set(args.require),
            set(args.forbid),
            args.max_tools,
            args.max_schema_chars,
            args.timeout,
        )
    )


if __name__ == "__main__":
    main()