Add everything-server for comprehensive MCP conformance testing

Implements a reference server that exercises all MCP protocol features
including tools, resources, prompts, logging, completion, sampling, and
elicitation. This server is used to validate the Python SDK against the
official MCP conformance test suite.

The implementation uses FastMCP for ergonomics and passes all 25
conformance scenarios. It matches the functionality of the TypeScript
everything-server to ensure consistency across language implementations.

Author: felixweinberger

examples/servers/everything-server/README.md

@@ -0,0 +1,42 @@
+# MCP Everything Server
+
+A comprehensive MCP server implementing all protocol features for conformance testing.
+
+## Overview
+
+The Everything Server is a reference implementation that demonstrates all features of the Model Context Protocol (MCP). It is designed to be used with the [MCP Conformance Test Framework](https://github.com/modelcontextprotocol/conformance) to validate MCP client and server implementations.
+
+## Installation
+
+From the python-sdk root directory:
+
+```bash
+uv sync --frozen
+```
+
+## Usage
+
+### Running the Server
+
+Start the server with default settings (port 3001):
+
+```bash
+uv run -m mcp_everything_server
+```
+
+Or with custom options:
+
+```bash
+uv run -m mcp_everything_server --port 3001 --log-level DEBUG
+```
+
+The server will be available at: `http://localhost:3001/mcp`
+
+### Command-Line Options
+
+- `--port` - Port to listen on (default: 3001)
+- `--log-level` - Logging level: DEBUG, INFO, WARNING, ERROR, CRITICAL (default: INFO)
+
+## Running Conformance Tests
+
+See the [MCP Conformance Test Framework](https://github.com/modelcontextprotocol/conformance) for instructions on running conformance tests against this server.

examples/servers/everything-server/mcp_everything_server/__init__.py

@@ -0,0 +1,3 @@
+"""MCP Everything Server - Comprehensive conformance test server."""
+
+__version__ = "0.1.0"

examples/servers/everything-server/mcp_everything_server/__main__.py

@@ -0,0 +1,6 @@
+"""CLI entry point for the MCP Everything Server."""
+
+from mcp_everything_server.server import main
+
+if __name__ == "__main__":
+    main()

examples/servers/everything-server/mcp_everything_server/server.py

@@ -0,0 +1,344 @@
+#!/usr/bin/env python3
+"""
+MCP Everything Server - Conformance Test Server
+
+Server implementing all MCP features for conformance testing based on Conformance Server Specification.
+"""
+
+import asyncio
+import logging
+from typing import Any
+
+import click
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.server.fastmcp.prompts.base import UserMessage
+from mcp.server.session import ServerSession
+from mcp.types import (
+    AudioContent,
+    CompletionArgument,
+    CompletionContext,
+    EmbeddedResource,
+    ImageContent,
+    PromptReference,
+    ResourceTemplateReference,
+    SamplingMessage,
+    TextContent,
+    TextResourceContents,
+)
+from pydantic import BaseModel, Field
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+# ===== TEST DATA =====
+
+# Sample base64 encoded 1x1 red PNG pixel for testing
+TEST_IMAGE_BASE64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
+
+# Sample base64 encoded minimal WAV file for testing
+TEST_AUDIO_BASE64 = "UklGRiYAAABXQVZFZm10IBAAAAABAAEAQB8AAAB9AAACABAAZGF0YQIAAAA="
+
+# ===== SERVER STATE =====
+
+# Resource subscriptions tracking
+resource_subscriptions: set[str] = set()
+watched_resource_content = "Watched resource content"
+
+# ===== CREATE FASTMCP SERVER =====
+
+mcp = FastMCP(
+    name="mcp-conformance-test-server",
+)
+
+
+# ===== TOOLS =====
+
+
+@mcp.tool()
+def test_simple_text() -> str:
+    """Tests simple text content response"""
+    return "This is a simple text response for testing."
+
+
+@mcp.tool()
+def test_image_content() -> list[ImageContent]:
+    """Tests image content response"""
+    return [ImageContent(type="image", data=TEST_IMAGE_BASE64, mimeType="image/png")]
+
+
+@mcp.tool()
+def test_audio_content() -> list[AudioContent]:
+    """Tests audio content response"""
+    return [AudioContent(type="audio", data=TEST_AUDIO_BASE64, mimeType="audio/wav")]
+
+
+@mcp.tool()
+def test_embedded_resource() -> list[EmbeddedResource]:
+    """Tests embedded resource content response"""
+    return [
+        EmbeddedResource(
+            type="resource",
+            resource=TextResourceContents(
+                uri="test://embedded-resource",
+                mimeType="text/plain",
+                text="This is an embedded resource content.",
+            ),
+        )
+    ]
+
+
+@mcp.tool()
+def test_multiple_content_types() -> list[TextContent | ImageContent | EmbeddedResource]:
+    """Tests response with multiple content types (text, image, resource)"""
+    return [
+        TextContent(type="text", text="Multiple content types test:"),
+        ImageContent(type="image", data=TEST_IMAGE_BASE64, mimeType="image/png"),
+        EmbeddedResource(
+            type="resource",
+            resource=TextResourceContents(
+                uri="test://mixed-content-resource",
+                mimeType="application/json",
+                text='{"test": "data", "value": 123}',
+            ),
+        ),
+    ]
+
+
+@mcp.tool()
+async def test_tool_with_logging(ctx: Context) -> str:
+    """Tests tool that emits log messages 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 "Tool with logging executed successfully"
+
+
+@mcp.tool()
+async def test_tool_with_progress(ctx: Context) -> str:
+    """Tests tool that reports progress notifications"""
+    await ctx.report_progress(progress=0, total=100, message="Completed step 0 of 100")
+    await asyncio.sleep(0.05)
+
+    await ctx.report_progress(progress=50, total=100, message="Completed step 50 of 100")
+    await asyncio.sleep(0.05)
+
+    await ctx.report_progress(progress=100, total=100, message="Completed step 100 of 100")
+
+    # Return progress token as string
+    progress_token = ctx.request_context.meta.progressToken if ctx.request_context and ctx.request_context.meta else 0
+    return str(progress_token)
+
+
+@mcp.tool()
+def test_error_handling() -> str:
+    """Tests error response handling"""
+    raise RuntimeError("This tool intentionally returns an error for testing")
+
+
+class SamplingInput(BaseModel):
+    prompt: str = Field(description="The prompt to send to the LLM")
+
+
+@mcp.tool()
+async def test_sampling(prompt: str, ctx: Context[ServerSession, None]) -> str:
+    """Tests server-initiated sampling (LLM completion request)"""
+    try:
+        # Request sampling from client
+        result = await ctx.session.create_message(
+            messages=[SamplingMessage(role="user", content=TextContent(type="text", text=prompt))],
+            max_tokens=100,
+        )
+
+        if result.content.type == "text":
+            model_response = result.content.text
+        else:
+            model_response = "No response"
+
+        return f"LLM response: {model_response}"
+    except Exception as e:
+        return f"Sampling not supported or error: {str(e)}"
+
+
+class ElicitationInput(BaseModel):
+    message: str = Field(description="The message to show the user")
+
+
+class UserResponse(BaseModel):
+    response: str = Field(description="User's response")
+
+
+@mcp.tool()
+async def test_elicitation(message: str, ctx: Context[ServerSession, None]) -> str:
+    """Tests server-initiated elicitation (user input request)"""
+    try:
+        # Request user input from client
+        result = await ctx.elicit(message=message, schema=UserResponse)
+
+        return (
+            f"User response: action={result.action}, content={result.data.model_dump_json() if result.data else '{}'}"
+        )
+    except Exception as e:
+        return f"Elicitation not supported or error: {str(e)}"
+
+
+# ===== RESOURCES =====
+
+
+@mcp.resource("test://static-text")
+def static_text_resource() -> str:
+    """A static text resource for testing"""
+    return "This is the content of the static text resource."
+
+
+@mcp.resource("test://static-binary")
+def static_binary_resource() -> bytes:
+    """A static binary resource (image) for testing"""
+    import base64
+
+    return base64.b64decode(TEST_IMAGE_BASE64)
+
+
+@mcp.resource("test://template/{id}/data")
+def template_resource(id: str) -> str:
+    """A resource template with parameter substitution"""
+    import json
+
+    return json.dumps({"id": id, "templateTest": True, "data": f"Data for ID: {id}"})
+
+
+@mcp.resource("test://watched-resource")
+def watched_resource() -> str:
+    """A resource that can be subscribed to for updates"""
+    return watched_resource_content
+
+
+# ===== PROMPTS =====
+
+
+@mcp.prompt()
+def test_simple_prompt() -> list[UserMessage]:
+    """A simple prompt without arguments"""
+    return [UserMessage(role="user", content=TextContent(type="text", text="This is a simple prompt for testing."))]
+
+
+@mcp.prompt()
+def test_prompt_with_arguments(arg1: str, arg2: str) -> list[UserMessage]:
+    """A prompt with required arguments"""
+    return [
+        UserMessage(
+            role="user", content=TextContent(type="text", text=f"Prompt with arguments: arg1='{arg1}', arg2='{arg2}'")
+        )
+    ]
+
+
+@mcp.prompt()
+def test_prompt_with_embedded_resource(resourceUri: str) -> list[UserMessage]:
+    """A prompt that includes an embedded resource"""
+    return [
+        UserMessage(
+            role="user",
+            content=EmbeddedResource(
+                type="resource",
+                resource=TextResourceContents(
+                    uri=resourceUri,
+                    mimeType="text/plain",
+                    text="Embedded resource content for testing.",
+                ),
+            ),
+        ),
+        UserMessage(role="user", content=TextContent(type="text", text="Please process the embedded resource above.")),
+    ]
+
+
+@mcp.prompt()
+def test_prompt_with_image() -> list[UserMessage]:
+    """A prompt that includes image content"""
+    return [
+        UserMessage(role="user", content=ImageContent(type="image", data=TEST_IMAGE_BASE64, mimeType="image/png")),
+        UserMessage(role="user", content=TextContent(type="text", text="Please analyze the image above.")),
+    ]
+
+
+# ===== CUSTOM REQUEST HANDLERS =====
+
+
+def setup_custom_handlers():
+    """Set up custom request handlers for logging and completion."""
+
+    # Handler for logging/setLevel
+    @mcp._mcp_server.set_logging_level()
+    async def handle_set_logging_level(level: str) -> None:
+        """Handle logging level changes"""
+        logger.info(f"Log level set to: {level}")
+        # In a real implementation, you would adjust the logging level here
+        # For conformance testing, we just acknowledge the request
+
+    # Handler for resources/subscribe
+    async def handle_subscribe(uri: str) -> dict[str, Any]:
+        """Handle resource subscription"""
+        resource_subscriptions.add(uri)
+        logger.info(f"Subscribed to resource: {uri}")
+        return {}
+
+    # Handler for resources/unsubscribe
+    async def handle_unsubscribe(uri: str) -> dict[str, Any]:
+        """Handle resource unsubscription"""
+        resource_subscriptions.discard(uri)
+        logger.info(f"Unsubscribed from resource: {uri}")
+        return {}
+
+    # Register subscription handlers
+    mcp._mcp_server.subscribe_resource()(handle_subscribe)
+    mcp._mcp_server.unsubscribe_resource()(handle_unsubscribe)
+
+    # Handler for completion/complete
+    @mcp._mcp_server.completion()
+    async def handle_completion(
+        ref: PromptReference | ResourceTemplateReference,
+        argument: CompletionArgument,
+        context: CompletionContext | None,
+    ) -> dict[str, Any]:
+        """Handle completion requests"""
+        # Basic completion support - returns empty array for conformance
+        # Real implementations would provide contextual suggestions
+        return {"values": [], "total": 0, "hasMore": False}
+
+
+# Set up custom handlers when module is loaded
+setup_custom_handlers()
+
+
+# ===== CLI =====
+
+
+@click.command()
+@click.option("--port", default=3001, help="Port to listen on for HTTP")
+@click.option(
+    "--log-level",
+    default="INFO",
+    help="Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)",
+)
+def main(port: int, log_level: str) -> int:
+    """Run the MCP Everything Server."""
+    # Configure logging
+    logging.basicConfig(
+        level=getattr(logging, log_level.upper()),
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
+
+    logger.info(f"Starting MCP Everything Server on port {port}")
+    logger.info(f"Endpoint will be: http://localhost:{port}/mcp")
+
+    # Update settings and run
+    mcp.settings.port = port
+    mcp.run(transport="streamable-http")
+
+    return 0
+
+
+if __name__ == "__main__":
+    main()