Run MCP conformance tests in CI (#3628)

🤖 Generated with Claude Code

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

.github/actions/run-pytest/action.yml

@@ -3,7 +3,7 @@ description: "Run pytest with appropriate flags for the test type and platform"
 
 inputs:
   test-type:
-    description: "Type of tests to run: unit, integration, or client_process"
+    description: "Type of tests to run: unit, integration, client_process, or conformance"
     required: false
     default: "unit"
 
@@ -23,8 +23,13 @@ runs:
           TIMEOUT="5"
           MAX_PROCS="0"
           EXTRA_FLAGS="-x"
+        elif [ "${{ inputs.test-type }}" == "conformance" ]; then
+          MARKER="conformance"
+          TIMEOUT="120"
+          MAX_PROCS="0"
+          EXTRA_FLAGS="-x"
         else
-          MARKER="not integration and not client_process"
+          MARKER="not integration and not client_process and not conformance"
           TIMEOUT="5"
           MAX_PROCS="4"
           EXTRA_FLAGS=""

.github/workflows/run-tests.yml

@@ -73,6 +73,29 @@ jobs:
         with:
           test-type: client_process
 
+  run_conformance_tests:
+    name: "MCP conformance tests"
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Setup uv
+        uses: ./.github/actions/setup-uv
+        with:
+          resolution: locked
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: "22"
+
+      - name: Run conformance tests
+        uses: ./.github/actions/run-pytest
+        with:
+          test-type: conformance
+
   run_integration_tests:
     name: "Integration tests"
     runs-on: ubuntu-latest

pyproject.toml

@@ -139,6 +139,7 @@ env = [
 markers = [
     "integration: marks tests as integration tests (deselect with '-m \"not integration\"')",
     "client_process: marks tests that spawn client processes via stdio transport. These can create issues when run in the same CI environment as other subprocess-based tests.",
+    "conformance: marks MCP conformance tests (require Node.js/npx)",
 ]
 # Automatically mark all tests in integration_tests folder
 pythonpath = ["."]

tests/conformance/expected-failures.yml

@@ -0,0 +1,6 @@
+server:
+  - completion-complete
+  - server-sse-polling
+  - resources-subscribe
+  - resources-unsubscribe
+  - dns-rebinding-protection

tests/conformance/server.py

@@ -0,0 +1,377 @@
+"""FastMCP conformance test server.
+
+Registers the exact tools, resources, and prompts expected by the
+MCP conformance test suite (https://github.com/modelcontextprotocol/conformance).
+"""
+
+import asyncio
+import base64
+import json
+import sys
+from enum import Enum as PyEnum
+
+import mcp.types
+from mcp.types import EmbeddedResource, ImageContent, TextContent
+from pydantic import AnyUrl, BaseModel, Field
+
+from fastmcp import FastMCP
+from fastmcp.exceptions import ToolError
+from fastmcp.prompts import Message
+from fastmcp.server.context import Context
+from fastmcp.tools.function_tool import FunctionTool
+from fastmcp.utilities.types import Audio, Image
+
+# Minimal 1x1 red PNG for image tests (89 bytes)
+_1X1_PNG = base64.b64decode(
+    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4"
+    "nGP4z8BQDwAEgAF/pooBPQAAAABJRU5ErkJggg=="
+)
+
+# Minimal valid WAV: 16-bit mono PCM, 44100 Hz, single silent sample
+_SILENT_WAV = (
+    b"RIFF"
+    + (38).to_bytes(4, "little")
+    + b"WAVEfmt "
+    + (16).to_bytes(4, "little")
+    + (1).to_bytes(2, "little")  # PCM
+    + (1).to_bytes(2, "little")  # mono
+    + (44100).to_bytes(4, "little")  # sample rate
+    + (88200).to_bytes(4, "little")  # byte rate
+    + (2).to_bytes(2, "little")  # block align
+    + (16).to_bytes(2, "little")  # bits per sample
+    + b"data"
+    + (2).to_bytes(4, "little")
+    + (0).to_bytes(2, "little")  # one silent sample
+)
+
+server = FastMCP("conformance-test-server", dereference_schemas=False)
+
+
+# ---------------------------------------------------------------------------
+# Tools
+# ---------------------------------------------------------------------------
+
+
+@server.tool(name="test_simple_text")
+async def test_simple_text() -> str:
+    """A simple text tool for conformance testing."""
+    return "This is a simple text response for testing."
+
+
+@server.tool(name="test_image_content")
+async def test_image_content() -> Image:
+    """Returns a PNG image."""
+    return Image(data=_1X1_PNG, format="png")
+
+
+@server.tool(name="test_audio_content")
+async def test_audio_content() -> Audio:
+    """Returns WAV audio."""
+    return Audio(data=_SILENT_WAV, format="wav")
+
+
+@server.tool(name="test_embedded_resource")
+async def test_embedded_resource() -> list:
+    """Returns an embedded resource."""
+    return [
+        EmbeddedResource(
+            type="resource",
+            resource=mcp.types.TextResourceContents(
+                uri=AnyUrl("test://embedded-resource"),
+                mimeType="text/plain",
+                text="This is an embedded resource content.",
+            ),
+        )
+    ]
+
+
+@server.tool(name="test_multiple_content_types")
+async def test_multiple_content_types() -> list:
+    """Returns mixed text, image, and resource content."""
+    return [
+        TextContent(type="text", text="This is a text part of the response."),
+        ImageContent(
+            type="image",
+            data=base64.b64encode(_1X1_PNG).decode(),
+            mimeType="image/png",
+        ),
+        EmbeddedResource(
+            type="resource",
+            resource=mcp.types.TextResourceContents(
+                uri=AnyUrl("test://mixed-content-resource"),
+                mimeType="application/json",
+                text='{"test":"data","value":123}',
+            ),
+        ),
+    ]
+
+
+@server.tool(name="test_error_handling")
+async def test_error_handling() -> str:
+    """Always returns an error."""
+    raise ToolError("This tool intentionally returns an error for testing")
+
+
+@server.tool(name="test_tool_with_logging")
+async def test_tool_with_logging(ctx: Context) -> str:
+    """Sends log notifications during execution."""
+    await ctx.info("Tool execution started")
+    await asyncio.sleep(0.05)
+    await ctx.info("Tool processing data")
+    await asyncio.sleep(0.05)
+    await ctx.info("Tool execution completed")
+    return "Logging test complete."
+
+
+@server.tool(name="test_tool_with_progress")
+async def test_tool_with_progress(ctx: Context) -> str:
+    """Reports progress notifications."""
+    await ctx.report_progress(0, 100)
+    await asyncio.sleep(0.05)
+    await ctx.report_progress(50, 100)
+    await asyncio.sleep(0.05)
+    await ctx.report_progress(100, 100)
+    return "Progress test complete."
+
+
+@server.tool(name="test_sampling")
+async def test_sampling(prompt: str, ctx: Context) -> str:
+    """Requests LLM sampling via the client."""
+    result = await ctx.sample(
+        messages=[prompt],
+        result_type=str,
+    )
+    return f"Sampling result: {result}"
+
+
+class _UserInfo(BaseModel):
+    username: str
+    email: str
+
+
+@server.tool(name="test_elicitation")
+async def test_elicitation(message: str, ctx: Context) -> str:
+    """Requests user input via elicitation."""
+    result = await ctx.elicit(message, _UserInfo)
+    return f"Elicitation result: {result}"
+
+
+class _UserStatus(str, PyEnum):
+    active = "active"
+    inactive = "inactive"
+    pending = "pending"
+
+
+class _DefaultsForm(BaseModel):
+    name: str = Field(default="John Doe", description="User name")
+    age: int = Field(default=30, description="User age")
+    score: float = Field(default=95.5, description="User score")
+    status: _UserStatus = Field(default=_UserStatus.active, description="User status")
+    verified: bool = Field(default=True, description="Verification status")
+
+
+@server.tool(name="test_elicitation_sep1034_defaults")
+async def test_elicitation_sep1034_defaults(ctx: Context) -> str:
+    """Tests elicitation with default values per SEP-1034."""
+    result = await ctx.elicit(
+        "Please review and update the form fields with defaults",
+        _DefaultsForm,
+    )
+    return f"Elicitation completed: {result}"
+
+
+@server.tool(name="test_elicitation_sep1330_enums")
+async def test_elicitation_sep1330_enums(ctx: Context) -> str:
+    """Tests elicitation with enum schema improvements per SEP-1330."""
+    result = await ctx.session.elicit(
+        message="Please select options from the enum fields",
+        requestedSchema={
+            "type": "object",
+            "properties": {
+                "untitledSingle": {
+                    "type": "string",
+                    "description": "Select one option",
+                    "enum": ["option1", "option2", "option3"],
+                },
+                "titledSingle": {
+                    "type": "string",
+                    "description": "Select one option with titles",
+                    "oneOf": [
+                        {"const": "value1", "title": "First Option"},
+                        {"const": "value2", "title": "Second Option"},
+                        {"const": "value3", "title": "Third Option"},
+                    ],
+                },
+                "legacyEnum": {
+                    "type": "string",
+                    "description": "Select one option (legacy)",
+                    "enum": ["opt1", "opt2", "opt3"],
+                    "enumNames": [
+                        "Option One",
+                        "Option Two",
+                        "Option Three",
+                    ],
+                },
+                "untitledMulti": {
+                    "type": "array",
+                    "description": "Select multiple options",
+                    "minItems": 1,
+                    "maxItems": 3,
+                    "items": {
+                        "type": "string",
+                        "enum": ["option1", "option2", "option3"],
+                    },
+                },
+                "titledMulti": {
+                    "type": "array",
+                    "description": "Select multiple options with titles",
+                    "minItems": 1,
+                    "maxItems": 3,
+                    "items": {
+                        "anyOf": [
+                            {"const": "value1", "title": "First Choice"},
+                            {"const": "value2", "title": "Second Choice"},
+                            {"const": "value3", "title": "Third Choice"},
+                        ]
+                    },
+                },
+            },
+            "required": [],
+        },
+        related_request_id=ctx.request_id,
+    )
+    return f"Elicitation completed: action={result.action}, content={json.dumps(result.content or {})}"
+
+
+async def _json_schema_2020_12_fn(
+    name: str | None = None,
+    address: dict | None = None,
+) -> str:
+    """Tool with JSON Schema 2020-12 features for conformance testing (SEP-1613)."""
+    return f"JSON Schema 2020-12 tool called with: name={name}, address={address}"
+
+
+server.add_tool(
+    FunctionTool(
+        fn=_json_schema_2020_12_fn,
+        name="json_schema_2020_12_tool",
+        description="Tool with JSON Schema 2020-12 features for conformance testing (SEP-1613)",
+        parameters={
+            "$schema": "https://json-schema.org/draft/2020-12/schema",
+            "type": "object",
+            "$defs": {
+                "address": {
+                    "type": "object",
+                    "properties": {
+                        "street": {"type": "string"},
+                        "city": {"type": "string"},
+                    },
+                }
+            },
+            "properties": {
+                "name": {"type": "string"},
+                "address": {"$ref": "#/$defs/address"},
+            },
+            "additionalProperties": False,
+        },
+    )
+)
+
+
+# ---------------------------------------------------------------------------
+# Resources
+# ---------------------------------------------------------------------------
+
+
+@server.resource(
+    "test://static-text",
+    name="Static text resource",
+    mime_type="text/plain",
+)
+async def static_text_resource() -> str:
+    """Returns static text content."""
+    return "This is the content of the static text resource."
+
+
+@server.resource(
+    "test://static-binary",
+    name="Static binary resource",
+    mime_type="image/png",
+)
+async def static_binary_resource() -> bytes:
+    """Returns a binary PNG image."""
+    return _1X1_PNG
+
+
+@server.resource(
+    "test://template/{id}/data",
+    name="Template resource",
+    mime_type="application/json",
+)
+async def template_resource(id: str) -> str:
+    """Returns JSON data with the template parameter substituted."""
+    return json.dumps({"id": id, "templateTest": True, "data": f"Data for ID: {id}"})
+
+
+@server.resource(
+    "test://watched-resource",
+    name="Watched resource",
+    mime_type="text/plain",
+)
+async def watched_resource() -> str:
+    """A resource that supports subscriptions."""
+    return "Watched resource content."
+
+
+# ---------------------------------------------------------------------------
+# Prompts
+# ---------------------------------------------------------------------------
+
+
+@server.prompt(name="test_simple_prompt")
+async def test_simple_prompt() -> str:
+    """A simple prompt for conformance testing."""
+    return "This is a simple prompt for testing."
+
+
+@server.prompt(name="test_prompt_with_arguments")
+async def test_prompt_with_arguments(arg1: str, arg2: str) -> str:
+    """A prompt that accepts arguments."""
+    return f"Prompt with arguments: arg1='{arg1}', arg2='{arg2}'"
+
+
+@server.prompt(name="test_prompt_with_embedded_resource")
+async def test_prompt_with_embedded_resource(resourceUri: str) -> list:
+    """A prompt that returns an embedded resource."""
+    return [
+        Message(
+            EmbeddedResource(
+                type="resource",
+                resource=mcp.types.TextResourceContents(
+                    uri=AnyUrl(resourceUri),
+                    mimeType="text/plain",
+                    text=f"Content of resource {resourceUri}",
+                ),
+            )
+        ),
+    ]
+
+
+@server.prompt(name="test_prompt_with_image")
+async def test_prompt_with_image() -> list:
+    """A prompt that returns an image."""
+    return [
+        Message(
+            ImageContent(
+                type="image",
+                data=base64.b64encode(_1X1_PNG).decode(),
+                mimeType="image/png",
+            )
+        ),
+        Message("Please analyze the image above."),
+    ]
+
+
+if __name__ == "__main__":
+    port = int(sys.argv[1]) if len(sys.argv) > 1 else 8000
+    server.run(transport="streamable-http", host="127.0.0.1", port=port)

tests/conformance/test_conformance.py

@@ -0,0 +1,98 @@
+"""Run the MCP conformance test suite against a FastMCP server.
+
+Requires Node.js and npx to be available on PATH.
+Mark: pytest -m conformance
+"""
+
+import shutil
+import socket
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+import pytest
+import uvicorn
+
+CONFORMANCE_DIR = Path(__file__).parent
+EXPECTED_FAILURES = CONFORMANCE_DIR / "expected-failures.yml"
+HOST = "127.0.0.1"
+MCP_PATH = "/mcp"
+
+
+def _get_free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("127.0.0.1", 0))
+        s.listen(1)
+        return s.getsockname()[1]
+
+
+@pytest.fixture(scope="module")
+def _require_npx():
+    if shutil.which("npx") is None:
+        pytest.skip("npx not found on PATH — install Node.js to run conformance tests")
+
+
+@pytest.fixture(scope="module")
+def conformance_server(_require_npx):
+    """Start the conformance test server in a background thread."""
+    from tests.conformance.server import server as mcp_server
+
+    port = _get_free_port()
+    app = mcp_server.http_app(transport="streamable-http", path=MCP_PATH)
+
+    config = uvicorn.Config(app, host=HOST, port=port, log_level="warning")
+    uv_server = uvicorn.Server(config)
+
+    thread = threading.Thread(target=uv_server.run, daemon=True)
+    thread.start()
+
+    # Wait for server to accept connections
+    url = f"http://{HOST}:{port}{MCP_PATH}"
+    deadline = time.monotonic() + 10
+    while time.monotonic() < deadline:
+        try:
+            with socket.create_connection((HOST, port), timeout=1):
+                break
+        except OSError:
+            time.sleep(0.1)
+    else:
+        pytest.fail("Conformance server did not start in time")
+
+    yield url
+
+    uv_server.should_exit = True
+    thread.join(timeout=5)
+
+
+@pytest.mark.conformance
+@pytest.mark.timeout(120)
+def test_mcp_conformance(conformance_server):
+    """Run the full MCP conformance test suite against the server."""
+    cmd = [
+        "npx",
+        "--yes",
+        "@modelcontextprotocol/conformance@latest",
+        "server",
+        "--url",
+        conformance_server,
+        "--suite",
+        "all",
+    ]
+
+    if EXPECTED_FAILURES.exists():
+        cmd.extend(["--expected-failures", str(EXPECTED_FAILURES)])
+
+    result = subprocess.run(cmd, capture_output=True, text=True, timeout=90)
+
+    # Print output for visibility in test results
+    if result.stdout:
+        print(result.stdout)
+    if result.stderr:
+        print(result.stderr)
+
+    assert result.returncode == 0, (
+        f"Conformance tests failed (exit code {result.returncode}).\n"
+        f"stdout:\n{result.stdout}\n"
+        f"stderr:\n{result.stderr}"
+    )