#!/usr/bin/env python3
"""Real local MCP stdio client/server smoke test, targeting the Python SDK 1.x line.

Run in an isolated Python environment with mcp>=1.28,<2 installed.
Place this file beside mcp_v1_bounded_server.py from the same article.
No cloud service, LLM, credential, GPU or unrestricted shell command is used.
"""
from __future__ import annotations

import argparse
import asyncio
from importlib.metadata import version
import json
from pathlib import Path
import sys

from mcp import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client


def _content_text(result: object) -> str:
    return "\n".join(getattr(c, "text", "") for c in result.content)


async def verify() -> dict[str, object]:
    sdk = version("mcp")
    if not sdk.startswith("1."):
        raise RuntimeError("This example requires the MCP Python SDK 1.x (pin mcp<2)")
    server = Path(__file__).with_name("mcp_v1_bounded_server.py")
    if not server.is_file():
        raise FileNotFoundError("Download the companion MCP v1 server into this folder")
    checks: list[str] = []
    params = StdioServerParameters(command=sys.executable, args=[str(server)])
    async with stdio_client(params) as (incoming, outgoing):
        async with ClientSession(incoming, outgoing) as session:
            await asyncio.wait_for(session.initialize(), 8.0)
            info = await asyncio.wait_for(session.list_tools(), 8.0)
            names = [tool.name for tool in info.tools]
            assert names == ["check_local_python"], names
            checks.append("stdio_initialize_and_tool_discovery")

            default = await asyncio.wait_for(
                session.call_tool("check_local_python", arguments={}), 8.0
            )
            assert not default.isError and _content_text(default).strip() == "python check OK"
            checks.append("default_tool_call")

            four = await asyncio.wait_for(
                session.call_tool("check_local_python", arguments={"timeout_seconds": 4}), 8.0
            )
            assert not four.isError and _content_text(four).strip() == "python check OK"
            checks.append("explicit_timeout_tool_call")

            invalid = await asyncio.wait_for(
                session.call_tool("check_local_python", arguments={"timeout_seconds": 0}), 8.0
            )
            assert not invalid.isError and "between 1 and 5" in _content_text(invalid)
            checks.append("invalid_parameter_rejected")

            retry = await asyncio.wait_for(
                session.call_tool("check_local_python", arguments={"timeout_seconds": 2}), 8.0
            )
            assert not retry.isError and _content_text(retry).strip() == "python check OK"
            checks.append("call_after_validation_error")

    return {
        "status": "PASS_REAL_MCP_V1_STDIO",
        "sdk_version": sdk,
        "transport": "stdio",
        "cases": checks,
        "count": len(checks),
        "actual_mcp_server_and_client_started": True,
        "real_llm_called": False,
        "external_api_calls": 0,
        "user_supplied_shell_commands": 0,
        "tested_v2_sdk": False,
        "test_scope": "Local fixed Python check tool, not a production deployment or long-running stress test.",
    }


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--self-test", action="store_true", required=True)
    ap.parse_args()
    try:
        result = asyncio.run(asyncio.wait_for(verify(), timeout=45))
        assert result["count"] == 5
        print(json.dumps(result, indent=2))
        return 0
    except (RuntimeError, FileNotFoundError, AssertionError, asyncio.TimeoutError):
        print(json.dumps({"status": "FAIL_LOCAL_CLIENT_SERVER_CHECK"}), file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())