How to Test an MCP Server: 3 Layers That Actually Work

"""Minimal two-tool MCP server used as the companion code for testing an
MCP server end-to-end.

Transports:
    stdio (default): the standard MCP stdio transport, driven by
        `mcp dev src/server.py` or by an MCP client/inspector spawning this
        file as a subprocess. This is the path exercised by
        scripts/inspector_checks.sh and tests/test_transport_stdio.py.

    http: a small stdlib-only HTTP + JSON-RPC-shaped transport with
        bearer-token (JWT) auth and scope-based tool visibility, used to
        demonstrate testing an MCP server's auth boundary (see
        tests/test_transport_http.py and tests/test_auth.py). This is not
        part of the official MCP SDK's transports; it is hand-rolled here
        purely for the auth/transport testing exercise in the companion
        blog post.

    sse: a companion stdlib-only Server-Sent Events transport demonstrating
        incremental delivery and Last-Event-ID resumption (see
        tests/test_transport_sse.py).

Usage:
    python src/server.py                                 # stdio (default)
    mcp dev src/server.py                                 # stdio via MCP Inspector dev UI
    python src/server.py --transport http --port 8787
    python src/server.py --transport sse --port 8787
"""

from __future__ import annotations

import argparse
import json
import sys
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any, Dict, FrozenSet, List

try:
    from .tools import create_ticket as _create_ticket, get_weather as _get_weather
    from .auth import AuthError, decode_bearer_token, require_scope
except ImportError:  # running as a top-level script: `python src/server.py`
    import os

    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    from tools import create_ticket as _create_ticket, get_weather as _get_weather
    from auth import AuthError, decode_bearer_token, require_scope

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("mcp-testing-demo")


@mcp.tool()
def get_weather(city: str) -> dict:
    """Get current weather conditions for a supported city.

    Args:
        city: City name, e.g. "Austin" or "Tokyo".
    """
    return _get_weather(city)


@mcp.tool()
def create_ticket(title: str, description: str = "") -> dict:
    """Create a support ticket and return its id, title, and status.

    Args:
        title: Short ticket title. Required, non-empty.
        description: Optional longer description of the issue.
    """
    return _create_ticket(title, description)


# ---------------------------------------------------------------------------
# Tool registry shared by the custom HTTP/SSE transports below. Kept separate
# from the FastMCP decorators above so the auth-aware transports can
# list/invoke tools without going through the stdio protocol.
# ---------------------------------------------------------------------------

_TOOL_SCHEMAS: Dict[str, Dict[str, Any]] = {
    "get_weather": {
        "name": "get_weather",
        "description": "Get current weather conditions for a supported city.",
        "inputSchema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
    "create_ticket": {
        "name": "create_ticket",
        "description": "Create a support ticket and return its id, title, and status.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "description": {"type": "string"},
            },
            "required": ["title"],
        },
    },
}

_TOOL_FUNCS = {
    "get_weather": lambda args: _get_weather(args["city"]),
    "create_ticket": lambda args: _create_ticket(
        args["title"], args.get("description", "")
    ),
}


def _call_tool(name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
    if name not in _TOOL_FUNCS:
        raise ValueError(f"unknown tool: {name}")
    result = _TOOL_FUNCS[name](arguments)
    return {"content": [{"type": "text", "text": json.dumps(result)}]}


def _tools_list_for_scope(granted: FrozenSet[str]) -> List[Dict[str, Any]]:
    return [schema for tname, schema in _TOOL_SCHEMAS.items() if tname in granted]


# ---------------------------------------------------------------------------
# HTTP transport: single JSON-RPC-shaped endpoint, bearer-token + scope auth.
# ---------------------------------------------------------------------------


class _HTTPHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, format: str, *args: Any) -> None:  # noqa: A002 - stdlib hook, quiet by default
        pass

    def _send_json(self, status: int, payload: Dict[str, Any]) -> None:
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self) -> None:  # noqa: N802 - stdlib naming convention
        content_length = int(self.headers.get("Content-Length", 0) or 0)
        raw_body = self.rfile.read(content_length) if content_length else b"{}"

        try:
            granted = decode_bearer_token(self.headers.get("Authorization"))
        except AuthError as exc:
            self._send_json(exc.status, {"error": str(exc)})
            return

        try:
            request = json.loads(raw_body or b"{}")
        except json.JSONDecodeError:
            self._send_json(400, {"error": "invalid JSON body"})
            return

        method = request.get("method")
        req_id = request.get("id")
        params = request.get("params") or {}

        if method == "tools/list":
            self._send_json(
                200,
                {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "result": {"tools": _tools_list_for_scope(granted)},
                },
            )
            return

        if method == "tools/call":
            name = params.get("name")
            try:
                require_scope(granted, name)
            except AuthError as exc:
                self._send_json(exc.status, {"error": str(exc)})
                return
            try:
                result = _call_tool(name, params.get("arguments") or {})
            except ValueError as exc:
                self._send_json(400, {"error": str(exc)})
                return
            self._send_json(200, {"jsonrpc": "2.0", "id": req_id, "result": result})
            return

        self._send_json(400, {"error": f"unsupported method: {method}"})


def _run_http(port: int) -> None:
    server = ThreadingHTTPServer(("127.0.0.1", port), _HTTPHandler)
    print(f"MCP HTTP transport listening on http://127.0.0.1:{port}", file=sys.stderr)
    server.serve_forever()


# ---------------------------------------------------------------------------
# SSE transport: deterministic 5-event stream with Last-Event-ID resumption.
# ---------------------------------------------------------------------------

_SSE_EVENTS: List[Dict[str, Any]] = [
    {
        "id": i,
        "event": "tool_status",
        "data": {"seq": i, "tool": "get_weather", "status": "ok"},
    }
    for i in range(1, 6)
]


class _SSEHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, format: str, *args: Any) -> None:  # noqa: A002
        pass

    def do_GET(self) -> None:  # noqa: N802
        try:
            decode_bearer_token(self.headers.get("Authorization"))
        except AuthError as exc:
            body = json.dumps({"error": str(exc)}).encode("utf-8")
            self.send_response(exc.status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return

        last_event_id_raw = self.headers.get("Last-Event-ID")
        last_event_id = int(last_event_id_raw) if last_event_id_raw else 0

        # No Content-Length and no chunked Transfer-Encoding on this response
        # (the whole point is to stream an open-ended sequence of frames), so
        # on an HTTP/1.1 connection the client has no way to know where the
        # response ends unless we close the socket ourselves. Force
        # Connection: close and tell BaseHTTPRequestHandler to close the
        # connection after this request instead of keeping it alive.
        self.close_connection = True
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache")
        self.send_header("Connection", "close")
        self.end_headers()

        for event in _SSE_EVENTS:
            if event["id"] <= last_event_id:
                continue  # already delivered before the reconnect; skip
            frame = (
                f"id: {event['id']}\n"
                f"event: {event['event']}\n"
                f"data: {json.dumps(event['data'])}\n\n"
            )
            try:
                self.wfile.write(frame.encode("utf-8"))
                self.wfile.flush()
            except (BrokenPipeError, ConnectionResetError):
                return  # client dropped mid-stream; nothing more to do
            time.sleep(0.05)  # force incremental delivery, one frame at a time


def _run_sse(port: int) -> None:
    server = ThreadingHTTPServer(("127.0.0.1", port), _SSEHandler)
    print(f"MCP SSE transport listening on http://127.0.0.1:{port}", file=sys.stderr)
    server.serve_forever()


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--transport",
        choices=["stdio", "http", "sse"],
        default="stdio",
        help="Transport to run (default: stdio).",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=8787,
        help="Port for the http/sse transports (default: 8787).",
    )
    args = parser.parse_args()

    if args.transport == "stdio":
        mcp.run()
    elif args.transport == "http":
        _run_http(args.port)
    else:
        _run_sse(args.port)


if __name__ == "__main__":
    main()

#!/usr/bin/env bash
#
# Inspector-CLI protocol checks for the MCP server in src/server.py.
#
# Drives the server over its default stdio transport via
# `@modelcontextprotocol/inspector --cli` (Node 18+, so `npx` is available)
# and asserts on the JSON-RPC responses with `jq`:
#
#   1. tools/list returns both tools (get_weather, create_ticket) with
#      non-empty schemas.
#   2. tools/call on get_weather for a known city returns content that
#      mentions "temperature".
#
# Exits non-zero on the first failed assertion.
#
# Prereqs: Node 18+, jq.
# Usage: bash scripts/inspector_checks.sh

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
SERVER_ENTRY="${REPO_ROOT}/src/server.py"
PYTHON_BIN="${PYTHON_BIN:-python3}"

fail() {
  echo "FAIL: $1" >&2
  exit 1
}

command -v npx >/dev/null 2>&1 || fail "npx not found (need Node 18+)"
command -v jq >/dev/null 2>&1 || fail "jq not found"
command -v "${PYTHON_BIN}" >/dev/null 2>&1 || fail "${PYTHON_BIN} not found"

inspector_cli() {
  npx --yes @modelcontextprotocol/inspector --cli "${PYTHON_BIN}" "${SERVER_ENTRY}" "$@"
}

echo "==> tools/list"
TOOLS_LIST_JSON="$(inspector_cli --method tools/list)"

echo "${TOOLS_LIST_JSON}" | jq -e '(.tools // .) | length >= 2' >/dev/null \
  || fail "tools/list did not return at least 2 tools"

echo "${TOOLS_LIST_JSON}" | jq -e '
  (.tools // .) as $tools
  | ($tools | any(.name == "get_weather"))
    and ($tools | any(.name == "create_ticket"))
' >/dev/null || fail "tools/list is missing get_weather and/or create_ticket"

echo "${TOOLS_LIST_JSON}" | jq -e '
  (.tools // .) as $tools
  | all($tools[]; .inputSchema != null and (.inputSchema | length) > 0)
' >/dev/null || fail "one or more tools has an empty inputSchema"

echo "OK: both tools present with non-empty schemas"

echo "==> tools/call get_weather (Austin)"
CALL_JSON="$(inspector_cli --method tools/call --tool-name get_weather --tool-arg city=Austin)"

echo "${CALL_JSON}" | jq -e '
  (.content // [])
  | any(.text // "" | test("temperature"; "i"))
' >/dev/null || fail "tools/call result content did not mention 'temperature'"

echo "OK: get_weather(Austin) response mentions 'temperature'"
echo "All inspector CLI checks passed."

"""Plain-Python tool implementations backing the MCP server.

These functions have no dependency on the `mcp` package, Claude, or any
LLM. They are imported directly by:

- `src/server.py`, which wraps them as MCP tools, and
- `tests/test_tools_unit.py`, which unit tests them with exact-match
  assertions (the fully deterministic layer of the testing pyramid
  described in the companion blog post).

Keeping this logic MCP-agnostic is what makes it cheaply and
deterministically testable: no protocol, no transport, no LLM in the
loop.
"""

from __future__ import annotations

import itertools
from typing import Dict

_KNOWN_CITIES: Dict[str, Dict[str, object]] = {
    "san francisco": {"temperature_f": 62, "conditions": "foggy"},
    "new york": {"temperature_f": 74, "conditions": "partly cloudy"},
    "austin": {"temperature_f": 91, "conditions": "sunny"},
    "london": {"temperature_f": 58, "conditions": "light rain"},
    "tokyo": {"temperature_f": 79, "conditions": "clear"},
}

_ticket_id_counter = itertools.count(1)


def get_weather(city: str) -> dict:
    """Return a deterministic weather reading for a known city.

    Args:
        city: City name, case-insensitive (e.g. "Austin", "san francisco").

    Returns:
        A dict with keys `city`, `temperature_f`, and `conditions`.

    Raises:
        ValueError: if `city` is empty/whitespace or not in the known
            city lookup table.
    """
    if not isinstance(city, str) or not city.strip():
        raise ValueError("city must be a non-empty string")

    key = city.strip().lower()
    if key not in _KNOWN_CITIES:
        raise ValueError(f"unknown city: {city!r}")

    reading = _KNOWN_CITIES[key]
    return {
        "city": city.strip(),
        "temperature_f": reading["temperature_f"],
        "conditions": reading["conditions"],
    }


def create_ticket(title: str, description: str = "") -> dict:
    """Create an in-memory support ticket.

    Args:
        title: Ticket title. Must be non-empty after stripping whitespace.
        description: Optional free-text description.

    Returns:
        A dict with keys `id`, `title`, `description`, and `status`.
        `status` is always `"open"` for newly created tickets. `id` is
        unique for the lifetime of the process.

    Raises:
        ValueError: if `title` is empty or only whitespace.
    """
    if not isinstance(title, str) or not title.strip():
        raise ValueError("title must be a non-empty string")

    ticket_id = f"TCKT-{next(_ticket_id_counter):04d}"
    return {
        "id": ticket_id,
        "title": title.strip(),
        "description": description.strip() if isinstance(description, str) else "",
        "status": "open",
    }

"""Deterministic unit tests for the plain-Python tool functions.

These import directly from src/tools.py — no MCP protocol, no transport, no
LLM in the loop — so every assertion here is an exact match. This is the
fastest and most reliable layer of the testing pyramid described in the
companion blog post; everything else in this repo builds on top of it.

Run: pytest tests/test_tools_unit.py -v
"""

from __future__ import annotations

import pytest

from src.tools import create_ticket, get_weather


class TestGetWeather:
    def test_known_city_returns_expected_shape(self) -> None:
        result = get_weather("Austin")
        assert result == {
            "city": "Austin",
            "temperature_f": 91,
            "conditions": "sunny",
        }

    def test_known_city_is_case_insensitive(self) -> None:
        result = get_weather("aUsTiN")
        assert result["city"] == "aUsTiN"
        assert result["temperature_f"] == 91
        assert result["conditions"] == "sunny"

    def test_known_city_strips_whitespace(self) -> None:
        result = get_weather("  Tokyo  ")
        assert result["city"] == "Tokyo"
        assert result["temperature_f"] == 79

    def test_unknown_city_raises_value_error(self) -> None:
        with pytest.raises(ValueError, match="unknown city"):
            get_weather("Atlantis")

    def test_empty_city_raises_value_error(self) -> None:
        with pytest.raises(ValueError, match="non-empty string"):
            get_weather("")

    def test_whitespace_only_city_raises_value_error(self) -> None:
        with pytest.raises(ValueError, match="non-empty string"):
            get_weather("   ")


class TestCreateTicket:
    def test_valid_ticket_has_expected_shape(self) -> None:
        ticket = create_ticket("Login button broken", "Clicking does nothing")
        assert ticket["title"] == "Login button broken"
        assert ticket["description"] == "Clicking does nothing"
        assert ticket["status"] == "open"
        assert ticket["id"].startswith("TCKT-")

    def test_description_defaults_to_empty_string(self) -> None:
        ticket = create_ticket("Some bug")
        assert ticket["description"] == ""

    def test_title_is_stripped(self) -> None:
        ticket = create_ticket("  Trailing space title  ")
        assert ticket["title"] == "Trailing space title"

    def test_empty_title_raises_value_error(self) -> None:
        with pytest.raises(ValueError, match="non-empty string"):
            create_ticket("")

    def test_whitespace_only_title_raises_value_error(self) -> None:
        with pytest.raises(ValueError, match="non-empty string"):
            create_ticket("   ")

    def test_ids_are_unique_across_calls(self) -> None:
        ids = {create_ticket(f"ticket {i}")["id"] for i in range(10)}
        assert len(ids) == 10

"""Tool-selection eval harness for the MCP tools in src/tools.py.

Unlike tests/test_tools_unit.py (exact-match, deterministic), this harness
measures whether an LLM reliably chooses the *correct* tool and reasonable
arguments for a natural-language prompt. LLM tool-calling is probabilistic,
so each scenario is run multiple times (N=15) and scored as a pass rate
rather than a single pass/fail. Scenarios with a pass rate below the
threshold are flagged as flaky.

Usage:
    export ANTHROPIC_API_KEY=sk-ant-...
    python eval/eval_tool_selection.py

Prereqs: pip install anthropic
"""

from __future__ import annotations

import os
import sys
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List

import anthropic

MODEL = "claude-opus-4-8"
RUNS_PER_SCENARIO = 15
PASS_RATE_THRESHOLD = 0.90

TOOLS: List[Dict[str, Any]] = [
    {
        "name": "get_weather",
        "description": "Get current weather conditions for a supported city.",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string", "description": "City name"}},
            "required": ["city"],
        },
    },
    {
        "name": "create_ticket",
        "description": "Create a support ticket for a reported issue.",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "Short ticket title"},
                "description": {"type": "string", "description": "Longer description"},
            },
            "required": ["title"],
        },
    },
]


@dataclass
class Scenario:
    name: str
    prompt: str
    expected_tool: str
    check_args: Callable[[Dict[str, Any]], bool]


def _nonempty(value: Any) -> bool:
    return isinstance(value, str) and len(value.strip()) > 0


SCENARIOS: List[Scenario] = [
    Scenario(
        name="weather_direct",
        prompt="What's the weather like in Austin right now?",
        expected_tool="get_weather",
        check_args=lambda args: "austin" in str(args.get("city", "")).lower(),
    ),
    Scenario(
        name="weather_indirect",
        prompt="I'm packing for a trip to Tokyo tomorrow, should I bring a jacket?",
        expected_tool="get_weather",
        check_args=lambda args: "tokyo" in str(args.get("city", "")).lower(),
    ),
    Scenario(
        name="ticket_bug_report",
        prompt=(
            "The checkout page throws a 500 error whenever I apply a discount "
            "code. Can you log this as an issue for the team?"
        ),
        expected_tool="create_ticket",
        check_args=lambda args: _nonempty(args.get("title")),
    ),
    Scenario(
        name="ticket_feature_request",
        prompt="Please file a ticket asking for dark mode support in the dashboard.",
        expected_tool="create_ticket",
        check_args=lambda args: _nonempty(args.get("title")),
    ),
]


@dataclass
class ScenarioResult:
    name: str
    passes: int = 0
    failures: List[str] = field(default_factory=list)

    @property
    def total(self) -> int:
        return self.passes + len(self.failures)

    @property
    def pass_rate(self) -> float:
        return self.passes / self.total if self.total else 0.0


def _run_once(client: "anthropic.Anthropic", scenario: Scenario) -> "str | None":
    """Return a failure reason string, or None if the scenario passed."""
    response = client.messages.create(
        model=MODEL,
        max_tokens=512,
        tools=TOOLS,
        messages=[{"role": "user", "content": scenario.prompt}],
    )

    tool_uses = [block for block in response.content if block.type == "tool_use"]
    if not tool_uses:
        return "no tool_use block in response"
    if len(tool_uses) > 1:
        return f"expected exactly one tool call, got {len(tool_uses)}"

    call = tool_uses[0]
    if call.name != scenario.expected_tool:
        return f"expected tool {scenario.expected_tool!r}, got {call.name!r}"

    if not scenario.check_args(call.input):
        return f"argument check failed for input {call.input!r}"

    return None


def main() -> int:
    if not os.environ.get("ANTHROPIC_API_KEY"):
        print("ERROR: ANTHROPIC_API_KEY is not set.", file=sys.stderr)
        return 2

    client = anthropic.Anthropic()
    results: List[ScenarioResult] = []

    for scenario in SCENARIOS:
        result = ScenarioResult(name=scenario.name)
        for _ in range(RUNS_PER_SCENARIO):
            failure = _run_once(client, scenario)
            if failure is None:
                result.passes += 1
            else:
                result.failures.append(failure)
        results.append(result)

        status = "OK" if result.pass_rate >= PASS_RATE_THRESHOLD else "FLAKY"
        print(
            f"[{status}] {scenario.name}: {result.passes}/{result.total} "
            f"({result.pass_rate:.0%})"
        )
        if status == "FLAKY":
            for reason in result.failures[:3]:
                print(f"    - {reason}")

    flaky = [r for r in results if r.pass_rate < PASS_RATE_THRESHOLD]
    print()
    if flaky:
        print(
            f"{len(flaky)}/{len(results)} scenario(s) below "
            f"{PASS_RATE_THRESHOLD:.0%} pass rate."
        )
        return 1

    print(f"All {len(results)} scenarios at or above {PASS_RATE_THRESHOLD:.0%} pass rate.")
    return 0


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

"""pytest suite exercising the server's default stdio transport end-to-end.

Spawns src/server.py as a stdio subprocess (the same way MCP Inspector and
any stdio-based MCP client would) and asserts:

  1. The initialize handshake completes within a timeout.
  2. A tools/call succeeds while the server is alive.
  3. After the client disconnects, the server process exits with code 0
     within a short timeout (no zombie process left behind).

Run: pytest tests/test_transport_stdio.py -v

Prereqs: pip install pytest mcp
"""

from __future__ import annotations

import asyncio
import os
import subprocess
import sys
import time

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

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SERVER_PATH = os.path.join(REPO_ROOT, "src", "server.py")
HANDSHAKE_TIMEOUT = 10.0
EXIT_TIMEOUT = 5.0


async def _initialize_and_call_tool():
    """Run the full stdio handshake + a tools/call, and return the result."""
    server_params = StdioServerParameters(command=sys.executable, args=[SERVER_PATH])
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await asyncio.wait_for(session.initialize(), timeout=HANDSHAKE_TIMEOUT)
            result = await asyncio.wait_for(
                session.call_tool("get_weather", {"city": "Austin"}),
                timeout=HANDSHAKE_TIMEOUT,
            )
            return result


class TestStdioHandshakeAndToolCall:
    def test_initialize_and_tool_call_succeed(self) -> None:
        result = asyncio.run(_initialize_and_call_tool())
        assert result is not None
        assert not getattr(result, "isError", False)

        texts = [block.text for block in result.content if block.type == "text"]
        assert any("temperature" in text for text in texts)


class TestStdioProcessExitsCleanly:
    """Verifies the server process exits (no zombie) once the client
    disconnects, using a directly-managed subprocess so the exit code is
    observable to this test.
    """

    @pytest.fixture
    def server_process(self):
        proc = subprocess.Popen(
            [sys.executable, SERVER_PATH],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        try:
            yield proc
        finally:
            # Guaranteed teardown even if the test body raises an assertion.
            if proc.poll() is None:
                proc.terminate()
                try:
                    proc.wait(timeout=EXIT_TIMEOUT)
                except subprocess.TimeoutExpired:
                    proc.kill()
                    proc.wait(timeout=EXIT_TIMEOUT)

    def test_process_exits_after_stdin_closed(self, server_process) -> None:
        # Give the server a moment to finish starting up before disconnecting.
        time.sleep(0.5)

        server_process.stdin.close()

        deadline = time.monotonic() + EXIT_TIMEOUT
        exit_code = None
        while time.monotonic() < deadline:
            exit_code = server_process.poll()
            if exit_code is not None:
                break
            time.sleep(0.1)

        assert exit_code is not None, "server process did not exit after stdin closed"
        assert exit_code == 0, f"server process exited with code {exit_code}"

"""pytest + httpx suite against the HTTP-transport instance of src/server.py.

This module starts `python src/server.py --transport http --port 8787`
itself (module-scoped fixture) so `pytest tests/test_transport_http.py -v`
works standalone with no manual setup step. Covers:

  - No Authorization header -> 401
  - Invalid bearer token -> 401
  - Valid token -> 200 + a well-formed tools/list
  - A second, narrower-scoped token -> tools/list containing only its
    scoped tools

The two bearer tokens below are HS256 JWTs pre-signed with the shared
secret in src/auth.py (`JWT_SECRET`), hardcoded here so this test file
doesn't need the `pyjwt` dependency (only pytest + httpx, per the spec).
Each has a far-future `exp` (year ~2099) so they never expire. See
tests/test_auth.py for the JWT-based expiry/refresh suite, which does use
pyjwt to mint tokens dynamically.

Run: pytest tests/test_transport_http.py -v

Prereqs: pip install pytest httpx
"""

from __future__ import annotations

import os
import subprocess
import sys
import time

import httpx
import pytest

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SERVER_PATH = os.path.join(REPO_ROOT, "src", "server.py")
PORT = 8787
BASE_URL = f"http://127.0.0.1:{PORT}"
STARTUP_TIMEOUT = 10.0

# Pre-signed HS256 JWTs (see module docstring). Scopes:
#   FULL_ACCESS_TOKEN    -> get_weather, create_ticket
#   WEATHER_ONLY_TOKEN   -> get_weather only
FULL_ACCESS_TOKEN = (
    "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
    "eyJzdWIiOiJmdWxsLWFjY2Vzcy10ZXN0Iiwic2NvcGUiOiJnZXRfd2VhdGhlciBjcmVhdGVfdGlja2V0IiwiZXhwIjo0MDg5MDAwMDAwfQ."
    "sfNcRB4Pgs6CYRNzxq5TtQeXgf-FjA7bgt6aY6F7baY"
)
WEATHER_ONLY_TOKEN = (
    "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
    "eyJzdWIiOiJ3ZWF0aGVyLW9ubHktdGVzdCIsInNjb3BlIjoiZ2V0X3dlYXRoZXIiLCJleHAiOjQwODkwMDAwMDB9."
    "mtms-f8F6v2KB-RfG-TsyCoNPWwx-1-s1CZMFPMTwP0"
)


@pytest.fixture(scope="module")
def http_server():
    proc = subprocess.Popen(
        [sys.executable, SERVER_PATH, "--transport", "http", "--port", str(PORT)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        deadline = time.monotonic() + STARTUP_TIMEOUT
        last_error: Exception | None = None
        while time.monotonic() < deadline:
            if proc.poll() is not None:
                stderr = proc.stderr.read().decode("utf-8", errors="replace")
                raise RuntimeError(f"server exited early:\n{stderr}")
            try:
                httpx.post(BASE_URL, json={}, timeout=0.5)
                break
            except httpx.TransportError as exc:
                last_error = exc
                time.sleep(0.1)
        else:
            raise RuntimeError(f"server did not start in time: {last_error}")

        yield BASE_URL
    finally:
        # Guaranteed teardown even if a test assertion fails.
        if proc.poll() is None:
            proc.terminate()
            try:
                proc.wait(timeout=5.0)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait(timeout=5.0)


def _rpc(base_url: str, method: str, params: dict, token: "str | None" = None):
    headers = {"Authorization": f"Bearer {token}"} if token else {}
    return httpx.post(
        base_url,
        json={"jsonrpc": "2.0", "id": 1, "method": method, "params": params},
        headers=headers,
        timeout=5.0,
    )


class TestHTTPAuth:
    def test_no_authorization_header_returns_401(self, http_server: str) -> None:
        response = _rpc(http_server, "tools/list", {})
        assert response.status_code == 401

    def test_invalid_bearer_token_returns_401(self, http_server: str) -> None:
        response = _rpc(http_server, "tools/list", {}, token="not-a-real-token")
        assert response.status_code == 401

    def test_valid_token_returns_well_formed_tools_list(self, http_server: str) -> None:
        response = _rpc(http_server, "tools/list", {}, token=FULL_ACCESS_TOKEN)
        assert response.status_code == 200

        body = response.json()
        tools = body["result"]["tools"]
        names = {tool["name"] for tool in tools}
        assert names == {"get_weather", "create_ticket"}
        for tool in tools:
            assert tool["description"]
            assert tool["inputSchema"]["type"] == "object"

    def test_scoped_token_only_lists_its_own_tools(self, http_server: str) -> None:
        response = _rpc(http_server, "tools/list", {}, token=WEATHER_ONLY_TOKEN)
        assert response.status_code == 200

        body = response.json()
        names = {tool["name"] for tool in body["result"]["tools"]}
        assert names == {"get_weather"}

"""pytest + httpx streaming suite against the SSE-transport instance of
src/server.py.

ASSUMPTION (documented up front, as this test relies on it): the SSE
transport's event sequence is a small, fixed, in-memory list
(`_SSE_EVENTS` in src/server.py) that is replayed identically on every
connection. Resumption via `Last-Event-ID` works by skipping events with
`id <= Last-Event-ID` from that same fixed sequence. This is a
teaching-scale stand-in for "server-side event-id persistence" -- a real
production server would persist delivered-event state per client/stream in
a durable store rather than replaying a hardcoded in-memory list, but the
resumption *contract* being tested (reconnect with Last-Event-ID; no
duplicate delivery of already-processed events) is the same either way.

This module starts `python src/server.py --transport sse --port 8787`
itself (module-scoped fixture) so `pytest tests/test_transport_sse.py -v`
works standalone with no manual setup step. Covers:

  - Incremental frame parsing: the client observes early events without
    waiting for the stream to close.
  - Mid-stream dropped connection: reconnecting with `Last-Event-ID`
    resumes from the next undelivered event and does not redeliver an
    already-processed event.

Run: pytest tests/test_transport_sse.py -v

Prereqs: pip install pytest httpx
"""

from __future__ import annotations

import os
import subprocess
import sys
import time
from typing import Iterator, Optional, Tuple

import httpx
import pytest

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SERVER_PATH = os.path.join(REPO_ROOT, "src", "server.py")
PORT = 8788
BASE_URL = f"http://127.0.0.1:{PORT}"
STARTUP_TIMEOUT = 10.0

# Same shared-secret HS256 JWT used in tests/test_transport_http.py, hardcoded
# here too so this file stays self-contained and doesn't need pyjwt. Grants
# both tool scopes; the SSE transport doesn't scope by tool, only by whether
# the token is valid at all.
FULL_ACCESS_TOKEN = (
    "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9."
    "eyJzdWIiOiJmdWxsLWFjY2Vzcy10ZXN0Iiwic2NvcGUiOiJnZXRfd2VhdGhlciBjcmVhdGVfdGlja2V0IiwiZXhwIjo0MDg5MDAwMDAwfQ."
    "sfNcRB4Pgs6CYRNzxq5TtQeXgf-FjA7bgt6aY6F7baY"
)


@pytest.fixture(scope="module")
def sse_server():
    proc = subprocess.Popen(
        [sys.executable, SERVER_PATH, "--transport", "sse", "--port", str(PORT)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        deadline = time.monotonic() + STARTUP_TIMEOUT
        last_error: Optional[Exception] = None
        while time.monotonic() < deadline:
            if proc.poll() is not None:
                stderr = proc.stderr.read().decode("utf-8", errors="replace")
                raise RuntimeError(f"server exited early:\n{stderr}")
            try:
                with httpx.stream(
                    "GET",
                    BASE_URL,
                    headers={"Authorization": f"Bearer {FULL_ACCESS_TOKEN}"},
                    timeout=0.5,
                ) as response:
                    response.read()
                break
            except httpx.TransportError as exc:
                last_error = exc
                time.sleep(0.1)
        else:
            raise RuntimeError(f"server did not start in time: {last_error}")

        yield BASE_URL
    finally:
        if proc.poll() is None:
            proc.terminate()
            try:
                proc.wait(timeout=5.0)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait(timeout=5.0)


def _iter_sse_events(response: httpx.Response) -> Iterator[Tuple[int, str, dict]]:
    """Parse `id:` / `event:` / `data:` frames, yielding (id, event, data)."""
    event_id: Optional[int] = None
    event_name: Optional[str] = None
    data_line: Optional[str] = None

    for line in response.iter_lines():
        if line == "":
            if event_id is not None and event_name is not None and data_line is not None:
                import json as _json

                yield event_id, event_name, _json.loads(data_line)
            event_id = event_name = data_line = None
            continue
        if line.startswith("id: "):
            event_id = int(line[len("id: ") :])
        elif line.startswith("event: "):
            event_name = line[len("event: ") :]
        elif line.startswith("data: "):
            data_line = line[len("data: ") :]


class TestSSEIncrementalDelivery:
    def test_first_event_observed_well_before_stream_closes(
        self, sse_server: str
    ) -> None:
        start = time.monotonic()
        time_to_first: Optional[float] = None
        received_ids = []

        with httpx.stream(
            "GET",
            sse_server,
            headers={"Authorization": f"Bearer {FULL_ACCESS_TOKEN}"},
            timeout=10.0,
        ) as response:
            assert response.status_code == 200
            for event_id, _event_name, _data in _iter_sse_events(response):
                if time_to_first is None:
                    time_to_first = time.monotonic() - start
                received_ids.append(event_id)

        total_time = time.monotonic() - start

        assert received_ids == [1, 2, 3, 4, 5]
        assert time_to_first is not None
        # The first event must arrive well before the full stream finishes --
        # proof the client parsed it incrementally rather than waiting for
        # the connection to close. The server paces frames ~0.05s apart, so
        # the first of five frames should land in well under half the total
        # stream duration.
        assert time_to_first < total_time * 0.6


class TestSSEReconnectResume:
    def test_reconnect_with_last_event_id_does_not_duplicate(
        self, sse_server: str
    ) -> None:
        received_before_drop = []

        # Connect, read the first two events, then abandon the connection
        # (simulating a mid-stream drop) before the server finishes sending.
        with httpx.stream(
            "GET",
            sse_server,
            headers={"Authorization": f"Bearer {FULL_ACCESS_TOKEN}"},
            timeout=10.0,
        ) as response:
            for event_id, _event_name, _data in _iter_sse_events(response):
                received_before_drop.append(event_id)
                if len(received_before_drop) == 2:
                    break  # drop the connection early

        assert received_before_drop == [1, 2]

        # Reconnect with Last-Event-ID set to the last event we actually
        # processed. The server must resume at the next event and must not
        # redeliver 1 or 2.
        received_after_reconnect = []
        with httpx.stream(
            "GET",
            sse_server,
            headers={
                "Authorization": f"Bearer {FULL_ACCESS_TOKEN}",
                "Last-Event-ID": str(received_before_drop[-1]),
            },
            timeout=10.0,
        ) as response:
            assert response.status_code == 200
            for event_id, _event_name, _data in _iter_sse_events(response):
                received_after_reconnect.append(event_id)

        assert received_after_reconnect == [3, 4, 5]
        assert set(received_before_drop).isdisjoint(received_after_reconnect)

"""pytest suite covering the HTTP-transport server's auth boundary in depth:
rejection, scope enforcement, and token expiry/refresh.

Shared test signing secret: this file signs its own JWTs with the literal
secret below, which MUST match `JWT_SECRET` in src/auth.py exactly --
that's what makes a token minted here acceptable to the server under test.
It's a plaintext demo secret on purpose (see src/auth.py's module
docstring); never reuse it, or this pattern, for a real deployment.

Run: pytest tests/test_auth.py -v

Prereqs: pip install pytest httpx pyjwt
"""

from __future__ import annotations

import os
import subprocess
import sys
import time
from typing import Optional

import httpx
import jwt
import pytest

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SERVER_PATH = os.path.join(REPO_ROOT, "src", "server.py")
PORT = 8789
BASE_URL = f"http://127.0.0.1:{PORT}"
STARTUP_TIMEOUT = 10.0

# Must match src/auth.py's JWT_SECRET / JWT_ALGORITHM exactly.
JWT_SECRET = "mcp-testing-demo-shared-secret-do-not-use-in-prod"
JWT_ALGORITHM = "HS256"

FAR_FUTURE_EXP = 4_089_000_000  # ~year 2099


def _make_token(scope: str, exp: float) -> str:
    return jwt.encode(
        {"sub": "test-token", "scope": scope, "exp": exp},
        JWT_SECRET,
        algorithm=JWT_ALGORITHM,
    )


@pytest.fixture(scope="module")
def auth_server():
    proc = subprocess.Popen(
        [sys.executable, SERVER_PATH, "--transport", "http", "--port", str(PORT)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    try:
        deadline = time.monotonic() + STARTUP_TIMEOUT
        last_error: Optional[Exception] = None
        while time.monotonic() < deadline:
            if proc.poll() is not None:
                stderr = proc.stderr.read().decode("utf-8", errors="replace")
                raise RuntimeError(f"server exited early:\n{stderr}")
            try:
                httpx.post(BASE_URL, json={}, timeout=0.5)
                break
            except httpx.TransportError as exc:
                last_error = exc
                time.sleep(0.1)
        else:
            raise RuntimeError(f"server did not start in time: {last_error}")

        yield BASE_URL
    finally:
        if proc.poll() is None:
            proc.terminate()
            try:
                proc.wait(timeout=5.0)
            except subprocess.TimeoutExpired:
                proc.kill()
                proc.wait(timeout=5.0)


def _call_tool(base_url: str, tool_name: str, arguments: dict, token: Optional[str], client: Optional[httpx.Client] = None):
    headers = {"Authorization": f"Bearer {token}"} if token else {}
    body = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {"name": tool_name, "arguments": arguments},
    }
    if client is not None:
        return client.post(base_url, json=body, headers=headers, timeout=5.0)
    return httpx.post(base_url, json=body, headers=headers, timeout=5.0)


class TestRejection:
    def test_no_token_is_rejected_with_no_tool_side_effect(self, auth_server: str) -> None:
        response = _call_tool(
            auth_server, "create_ticket", {"title": "should not be created"}, token=None
        )
        assert response.status_code in (401, 403)
        body = response.json()
        assert "result" not in body

    def test_invalid_token_is_rejected_with_no_tool_side_effect(self, auth_server: str) -> None:
        response = _call_tool(
            auth_server,
            "create_ticket",
            {"title": "should not be created either"},
            token="garbage.not.a.jwt",
        )
        assert response.status_code in (401, 403)
        body = response.json()
        assert "result" not in body


class TestScoping:
    def test_weather_only_token_succeeds_on_get_weather(self, auth_server: str) -> None:
        token = _make_token(scope="get_weather", exp=FAR_FUTURE_EXP)
        response = _call_tool(auth_server, "get_weather", {"city": "London"}, token=token)
        assert response.status_code == 200
        assert response.json()["result"] is not None

    def test_weather_only_token_gets_403_on_create_ticket(self, auth_server: str) -> None:
        token = _make_token(scope="get_weather", exp=FAR_FUTURE_EXP)
        response = _call_tool(
            auth_server, "create_ticket", {"title": "nope"}, token=token
        )
        assert response.status_code == 403

    def test_ticket_only_token_succeeds_on_create_ticket(self, auth_server: str) -> None:
        token = _make_token(scope="create_ticket", exp=FAR_FUTURE_EXP)
        response = _call_tool(
            auth_server, "create_ticket", {"title": "filed via scoped token"}, token=token
        )
        assert response.status_code == 200
        assert response.json()["result"] is not None

    def test_ticket_only_token_gets_403_on_get_weather(self, auth_server: str) -> None:
        token = _make_token(scope="create_ticket", exp=FAR_FUTURE_EXP)
        response = _call_tool(auth_server, "get_weather", {"city": "London"}, token=token)
        assert response.status_code == 403


class TestExpiryAndRefresh:
    def test_expired_token_is_rejected(self, auth_server: str) -> None:
        expired_token = _make_token(scope="get_weather", exp=time.time() - 100)
        response = _call_tool(
            auth_server, "get_weather", {"city": "London"}, token=expired_token
        )
        assert response.status_code == 401
        assert "expired" in response.json()["error"].lower()

    def test_fresh_replacement_token_succeeds_without_new_connection(
        self, auth_server: str
    ) -> None:
        expired_token = _make_token(scope="get_weather", exp=time.time() - 100)

        with httpx.Client() as client:
            rejected = _call_tool(
                auth_server, "get_weather", {"city": "London"}, token=expired_token, client=client
            )
            assert rejected.status_code == 401

            # Same httpx.Client (and therefore, potentially, the same
            # underlying TCP connection) -- only the bearer token changes.
            fresh_token = _make_token(scope="get_weather", exp=FAR_FUTURE_EXP)
            accepted = _call_tool(
                auth_server, "get_weather", {"city": "London"}, token=fresh_token, client=client
            )
            assert accepted.status_code == 200
            assert accepted.json()["result"] is not None