[minimcp] Add integration tests for MCP servers built with different transports

- Add integration test suite for StdioTransport
- Add integration test suite for HTTPTransport
- Add integration test suite for StreamableHTTPTransport
- Add test helpers (client session, HTTP utilities, process management)
- Add math_mcp example server for integration testing
- Add server fixtures and conftest configuration
- Add psutil dependency for process management in tests

Author: sreenaths

pyproject.toml

@@ -63,6 +63,7 @@ dev = [
     "inline-snapshot>=0.23.0",
     "dirty-equals>=0.9.0",
     "coverage[toml]==7.10.7",
+    "psutil>=7.1.3",
 ]
 docs = [
     "mkdocs>=1.6.1",

tests/conftest.py

@@ -4,3 +4,12 @@
 @pytest.fixture
 def anyio_backend():
     return "asyncio"
+
+
+def pytest_addoption(parser: pytest.Parser) -> None:
+    parser.addoption(
+        "--use-existing-minimcp-server",
+        action="store_true",
+        default=False,
+        help="Use an already running MinimCP server if available.",
+    )

tests/server/minimcp/integration/conftest.py

@@ -0,0 +1,56 @@
+"""
+Shared fixtures for MCP integration tests.
+"""
+
+from collections.abc import AsyncGenerator
+
+import anyio
+import pytest
+import servers.http_server as http_test_server
+from helpers.http import until_available, url_available
+from helpers.process import run_module
+from servers.http_server import SERVER_HOST, SERVER_PORT
+
+pytestmark = pytest.mark.anyio
+
+BASE_URL: str = f"http://{SERVER_HOST}:{SERVER_PORT}"
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    use_existing_minimcp_server = config.getoption("--use-existing-minimcp-server")
+    server_is_running = anyio.run(url_available, BASE_URL)
+
+    if server_is_running and not use_existing_minimcp_server:
+        raise RuntimeError(
+            f"Server is already running at {BASE_URL}. "
+            "Run pytest with the --use-existing-minimcp-server option to use it."
+        )
+
+
+@pytest.fixture(scope="session")
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.fixture(scope="session")
+async def http_test_server_process() -> AsyncGenerator[None, None]:
+    """
+    Session-scoped fixture that starts the HTTP test server once across all workers.
+
+    With pytest-xdist, multiple workers may call this fixture. The first worker starts the server,
+    and subsequent workers detect and reuse it.
+    """
+
+    if await url_available(BASE_URL):
+        # Server is available, use that.
+        yield None
+    else:
+        try:
+            async with run_module(http_test_server):
+                await until_available(BASE_URL)
+                yield None
+                await anyio.sleep(1)  # Wait a bit for safe shutdown
+        except Exception:
+            # If server started between our check and start attempt, that's OK
+            # Another worker got there first
+            yield None

tests/server/minimcp/integration/helpers/client_session_with_init.py

@@ -0,0 +1,7 @@
+from mcp import ClientSession, InitializeResult
+
+
+class ClientSessionWithInit(ClientSession):
+    """A client session that stores the initialization result."""
+
+    initialize_result: InitializeResult | None = None

tests/server/minimcp/integration/helpers/http.py

@@ -0,0 +1,32 @@
+import anyio
+import httpx
+
+
+async def url_available(url: str) -> bool:
+    """Check if a URL is available (server is responding).
+
+    Returns True if the server responds with any status code (including 405 Method Not Allowed),
+    which indicates the server is running. Returns False only on connection errors.
+    """
+    try:
+        async with httpx.AsyncClient(follow_redirects=True) as client:
+            await client.get(url, timeout=2.0)
+            # Any response (including 405, 404, etc.) means the server is running
+            return True
+    except Exception:
+        # Connection refused, timeout, etc. - server not available
+        return False
+
+
+async def until_available(url: str, max_attempts: int = 60, sleep_interval: float = 0.5) -> None:
+    """Wait for a URL to become available.
+
+    Default timeout is 30 seconds (60 attempts * 0.5 seconds).
+    This gives the server enough time to start even under heavy system load.
+    """
+    for _ in range(max_attempts):
+        if await url_available(url):
+            return None
+        await anyio.sleep(sleep_interval)
+
+    raise RuntimeError(f"URL {url} is not available after {max_attempts * sleep_interval} seconds")

tests/server/minimcp/integration/helpers/process.py

@@ -0,0 +1,64 @@
+import os
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+from pathlib import Path
+from subprocess import DEVNULL, Popen
+from types import ModuleType
+
+from psutil import NoSuchProcess, Process, ZombieProcess, process_iter  # type: ignore
+
+
+def find_process(cmd_substr: str) -> Process | None:
+    """Find a process by file_name"""
+    try:
+        for proc in process_iter(["pid", "name", "cmdline"]):
+            cmdline = proc.info.get("cmdline", [])
+            if cmdline and any(cmd_substr in str(cmd) for cmd in cmdline):
+                return Process(proc.info["pid"])
+    except (NoSuchProcess, ZombieProcess):
+        pass
+
+    return None
+
+
+@asynccontextmanager
+async def run_subprocess(cmd: list[str], cwd: Path) -> AsyncGenerator[Process, None]:
+    """Run a subprocess and yield the process."""
+    sub_proc = None
+
+    try:
+        sub_proc = Popen(
+            cmd,
+            stdout=DEVNULL,
+            stderr=DEVNULL,
+            text=True,
+            cwd=cwd,
+        )
+
+        process = Process(sub_proc.pid)
+
+        if not process.is_running():
+            raise RuntimeError(f"Process for command {cmd} exited unexpectedly.")
+
+        yield process
+
+    finally:
+        if sub_proc and sub_proc.poll() is None:
+            sub_proc.terminate()
+            sub_proc.wait(5)
+            if sub_proc.poll() is None:
+                raise RuntimeError("Process was not terminated.")
+
+
+@asynccontextmanager
+async def run_module(module: ModuleType) -> AsyncGenerator[Process, None]:
+    """Run a module as a subprocess and yield the process."""
+
+    project_root = Path(__file__).parent.parent.parent.parent.parent.parent
+    path_relative_to_root = Path(module.__file__ or "").relative_to(project_root)
+    module_name = str(path_relative_to_root.with_suffix("")).replace(os.sep, ".")
+
+    cmd = ["uv", "run", "python", "-m", module_name]
+
+    async with run_subprocess(cmd, project_root) as process:
+        yield process

tests/server/minimcp/integration/servers/http_server.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""
+Test server for HTTP transport integration tests.
+"""
+
+import os
+
+import uvicorn
+from starlette.applications import Starlette
+
+from mcp.server.minimcp import HTTPTransport, StreamableHTTPTransport
+
+from .math_mcp import math_mcp
+
+SERVER_HOST = os.environ.get("TEST_SERVER_HOST", "127.0.0.1")
+SERVER_PORT = int(os.environ.get("TEST_SERVER_PORT", "30789"))
+
+HTTP_MCP_PATH = "/mcp/"
+STREAMABLE_HTTP_MCP_PATH = "/streamable-mcp/"
+
+
+def main():
+    """Main entry point for the test server."""
+
+    http_transport = HTTPTransport[None](math_mcp)
+    streamable_http_transport = StreamableHTTPTransport[None](math_mcp)
+
+    # In Starlette, the lifespan events do not run for mounted sub-applications,
+    # and this is expected behavior. Hence adding manually.
+    app = Starlette(lifespan=streamable_http_transport.lifespan)
+
+    app.mount(HTTP_MCP_PATH, http_transport.as_starlette())
+    app.mount(STREAMABLE_HTTP_MCP_PATH, streamable_http_transport.as_starlette())
+
+    uvicorn.run(
+        app,
+        host=SERVER_HOST,
+        port=SERVER_PORT,
+        log_level="info",
+        access_log=False,
+    )
+
+
+if __name__ == "__main__":
+    main()

tests/server/minimcp/integration/servers/math_mcp.py

@@ -0,0 +1,120 @@
+"""
+MiniMCP math server for integration tests.
+"""
+
+from typing import Any
+
+import anyio
+from pydantic import Field
+
+from mcp.server.minimcp import MiniMCP
+
+MATH_CONSTANTS = {
+    "pi": 3.14159265359,
+    "𝜋": 3.14159265359,
+    "e": 2.71828182846,
+    "𝚎": 2.71828182846,
+    "golden_ratio": 1.61803398875,
+    "𝚽": 1.61803398875,
+    "sqrt_2": 1.41421356237,
+    "√2": 1.41421356237,
+}
+
+
+# Create a simple math server for testing directly in this file
+math_mcp = MiniMCP[Any](
+    name="TestMathServer",
+    version="0.1.0",
+    instructions="A simple MCP server for mathematical operations used in integration tests.",
+    include_stack_trace=True,
+)
+
+
+# -- Tools --
+@math_mcp.tool()
+def add(a: float = Field(description="The first number"), b: float = Field(description="The second number")) -> float:
+    """Add two numbers"""
+    return a + b
+
+
+@math_mcp.tool()
+def add_with_const(
+    a: float | str = Field(description="The first value"), b: float | str = Field(description="The second value")
+) -> float:
+    """Add two values. Values can be numbers or mathematical constants."""
+    if isinstance(a, str):
+        a = float(MATH_CONSTANTS[a])
+    if isinstance(b, str):
+        b = float(MATH_CONSTANTS[b])
+    return a + b
+
+
+@math_mcp.tool()
+def subtract(
+    a: float = Field(description="The first number"), b: float = Field(description="The second number")
+) -> float:
+    """Subtract two numbers"""
+    return a - b
+
+
+@math_mcp.tool()
+def multiply(
+    a: float = Field(description="The first number"), b: float = Field(description="The second number")
+) -> float:
+    """Multiply two numbers"""
+    return a * b
+
+
+@math_mcp.tool()
+def divide(
+    a: float = Field(description="The first number"), b: float = Field(description="The second number")
+) -> float:
+    """Divide two numbers"""
+    if b == 0:
+        raise ValueError("Cannot divide by zero")
+    return a / b
+
+
+@math_mcp.tool(description="Add two numbers")
+async def add_with_progress(
+    a: float = Field(description="The first float number"), b: float = Field(description="The second float number")
+) -> float:
+    responder = math_mcp.context.get_responder()
+    await responder.report_progress(0.1, message="Adding numbers")
+    await anyio.sleep(0.1)
+    await responder.report_progress(0.4, message="Adding numbers")
+    await anyio.sleep(0.1)
+    await responder.report_progress(0.7, message="Adding numbers")
+    await anyio.sleep(0.1)
+    return a + b
+
+
+# -- Prompts --
+@math_mcp.prompt()
+def math_help(operation: str = Field(description="The mathematical operation to get help with")) -> str:
+    """Get help with mathematical operations"""
+    return f"""You are a helpful math assistant.
+Provide guidance on how to perform the following mathematical operation: {operation}
+
+Include:
+1. Step-by-step instructions
+2. Example calculations
+3. Common pitfalls to avoid
+"""
+
+
+# -- Resources --
+
+
+@math_mcp.resource("math://constants")
+def get_math_constants() -> dict[str, float]:
+    """Mathematical constants reference"""
+    return MATH_CONSTANTS
+
+
+@math_mcp.resource("math://constants/{constant_name}")
+def get_math_constant(constant_name: str) -> float:
+    """Get a specific mathematical constant"""
+    if constant_name not in MATH_CONSTANTS:
+        raise ValueError(f"Unknown constant: {constant_name}")
+    return MATH_CONSTANTS[constant_name]

tests/server/minimcp/integration/servers/stdio_server.py

@@ -0,0 +1,31 @@
+import logging
+import sys
+
+import anyio
+
+from mcp.server.minimcp import StdioTransport
+
+from .math_mcp import math_mcp
+
+# Configure logging for the test server
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    handlers=[
+        logging.StreamHandler(sys.stderr),  # Log to stderr to avoid interfering with stdio transport
+    ],
+)
+logger = logging.getLogger(__name__)
+
+
+def main():
+    """Main entry point for the test math server"""
+
+    logger.info("Test MiniMCP: Starting stdio server, listening for messages...")
+
+    transport = StdioTransport[None](math_mcp)
+    anyio.run(transport.run)
+
+
+if __name__ == "__main__":
+    main()