github · python-sdke8e64842781c

Evidence

# Try both npx.cmd and npx.exe on Windows

        for cmd in ["npx.cmd", "npx.exe", "npx"]:

            try:

                subprocess.run([cmd, "--version"], check=True, capture_output=True, shell=True)

                return cmd

            except subprocess.CalledProcessError:

                continue

RemediationAI

The problem is that `subprocess.run()` is called with `shell=True` while executing a command list, which is unnecessary and dangerous. Remove the `shell=True` parameter from the `subprocess.run()` call in `src/mcp/cli/cli.py` — the function already passes a list of arguments `[cmd, "--version"]`, so shell interpretation is not needed. With `shell=False` (the default), the subprocess module will execute the command directly without invoking a shell, eliminating command injection risks. Verify the fix by running the CLI and confirming that NPX version detection still works: `python -m mcp.cli.cli --help` should execute without shell-related errors.

Evidence

"""MCP unified conformance test client.

This client is designed to work with the @modelcontextprotocol/conformance npm package.

It handles all conformance test scenarios via environment variables and CLI arguments.

Contract:

    - MCP_CONFORMANCE_SCENARIO env var -> scenario name

    - MCP_CONFORMANCE_CONTEXT env var -> optional JSON (for client-credentials scenarios)

    - Server URL as last CLI argument (sys.argv[1])

    - Must exit 0 within 30 seconds

Scenarios:

    initialize             

RemediationAI

The problem is that the conformance test client uses a decorator-based handler registration pattern (`@register()`) that allows runtime re-registration of tool behavior, creating a vector for handlers to modify their own behavior or be overwritten by malicious test scenarios. Implement immutable handler registration by removing the `register()` function or converting `HANDLERS` from a mutable dict to a frozen mapping (e.g., `types.MappingProxyType`) in `.github/actions/conformance/client.py`. This prevents any scenario from re-registering or modifying handler behavior after initial setup. Verify by attempting to re-register a handler after the client starts — the operation should raise an `AttributeError` or `TypeError` indicating the mapping is read-only.

Evidence

# MCP Python SDK

<div align="center">

<strong>Python implementation of the Model Context Protocol (MCP)</strong>

[![PyPI][pypi-badge]][pypi-url]

[![MIT licensed][mit-badge]][mit-url]

[![Python Version][python-badge]][python-url]

[![Documentation][docs-badge]][docs-url]

[![Protocol][protocol-badge]][protocol-url]

[![Specification][spec-badge]][spec-url]

</div>

<!-- TODO(v2): Replace this README with README.v2.md when v2 is released -->

> [!NOTE]

> **This README documents v1.x of the MCP Pyt

RemediationAI

The problem is that README.md documents MCP resource handlers without mentioning path canonicalization, which could lead developers to implement handlers that accept file URIs without validation. Add a security section to README.md that explicitly warns developers to canonicalize all file paths using `os.path.realpath()` combined with `os.path.commonpath()` to verify the resolved path stays within an allowed root directory before opening files. Include a code example showing the safe pattern. Verify by creating a test resource handler that attempts to access `../../../etc/passwd` — the canonicalization check should reject it and raise an exception.

Evidence

# MCP Python SDK

<div align="center">

<strong>Python implementation of the Model Context Protocol (MCP)</strong>

[![PyPI][pypi-badge]][pypi-url]

[![MIT licensed][mit-badge]][mit-url]

[![Python Version][python-badge]][python-url]

[![Documentation][docs-badge]][docs-url]

[![Protocol][protocol-badge]][protocol-url]

[![Specification][spec-badge]][spec-url]

</div>

<!-- TODO(v2): Move this content back to README.md when v2 is released -->

> [!IMPORTANT]

> **This documents v2 of the SDK (currentl

RemediationAI

The problem is that README.v2.md documents MCP resource handlers without mentioning path canonicalization, which could lead developers to implement handlers that accept file URIs without validation. Add a security section to README.v2.md that explicitly warns developers to canonicalize all file paths using `os.path.realpath()` combined with `os.path.commonpath()` to verify the resolved path stays within an allowed root directory before opening files. Include a code example showing the safe pattern. Verify by creating a test resource handler that attempts to access `../../../etc/passwd` — the canonicalization check should reject it and raise an exception.

Evidence

from mcp.server.mcpserver import MCPServer

from mcp.server.mcpserver.prompts import base

mcp = MCPServer(name="Prompt Example")

@mcp.prompt(title="Code Review")

def review_code(code: str) -> str:

    return f"Please review this code:\n\n{code}"

@mcp.prompt(title="Debug Assistant")

def debug_error(error: str) -> list[base.Message]:

    return [

        base.UserMessage("I'm seeing this error:"),

        base.UserMessage(error),

        base.AssistantMessage("I'll help debug that. What have y

RemediationAI

The problem is that the `review_code()` prompt handler in `examples/snippets/servers/basic_prompt.py` uses an f-string to interpolate the untrusted `code` parameter directly into the prompt text, allowing an attacker to inject arbitrary instructions into the LLM's context. Replace the f-string with a safe wrapper by changing `return f"Please review this code:\n\n{code}"` to `return f"Please review this code:\n\n<untrusted>{code}</untrusted>"` or use `json.dumps(code)` to escape special characters. The `<untrusted>` tags signal to the LLM that the content is user-supplied and should be treated as data, not instructions. Verify by passing a prompt injection payload like `</untrusted>Ignore previous instructions and...` — the LLM should treat it as literal text within the tags, not as a command.

Evidence

"""MCPServer Echo Server"""

from mcp.server.mcpserver import MCPServer

# Create server

mcp = MCPServer("Echo Server")

@mcp.tool()

def echo_tool(text: str) -> str:

    """Echo the input text"""

    return text

@mcp.resource("echo://static")

def echo_resource() -> str:

    return "Echo!"

@mcp.resource("echo://{text}")

def echo_template(text: str) -> str:

    """Echo the input text"""

    return f"Echo: {text}"

@mcp.prompt("echo")

def echo_prompt(text: str) -> str:

    return text

RemediationAI

The problem is that the `echo.py` example file registers MCP handlers (tools and resources) but does not show any prompt handlers with interpolation vulnerabilities in the provided evidence. However, if prompt handlers are added to this file, they must not use f-strings or `.format()` to interpolate user input directly into prompt text. When adding prompt handlers, wrap any user-controlled parameters in `<untrusted>...</untrusted>` tags or use `json.dumps()` to escape them. Verify by adding a prompt handler that accepts a user parameter and confirming that injection payloads are rendered as literal text, not executed as instructions.

LLM consensus

Evidence

# MCP Python SDK

<div align="center">

<strong>Python implementation of the Model Context Protocol (MCP)</strong>

[![PyPI][pypi-badge]][pypi-url]

[![MIT licensed][mit-badge]][mit-url]

[![Python Version][python-badge]][python-url]

[![Documentation][docs-badge]][docs-url]

[![Protocol][protocol-badge]][protocol-url]

[![Specification][spec-badge]][spec-url]

</div>

<!-- TODO(v2): Replace this README with README.v2.md when v2 is released -->

> [!NOTE]

> **This README documents v1.x of the MCP Pyt

RemediationAI

The problem is that README.md documents MCP prompt handlers without explicitly warning against prompt injection via string interpolation. Add a security best-practices section to README.md that prohibits f-strings, template literals, and `.format()` calls on user input in prompt handlers, and instead mandates wrapping untrusted parameters in `<untrusted>...</untrusted>` tags or using `json.dumps()`. Include a code example showing the vulnerable vs. safe pattern. Verify by reviewing the documentation and confirming that all prompt handler examples use safe escaping techniques.

Evidence

"""MCPServer quickstart example.

Run from the repository root:

    uv run examples/snippets/servers/mcpserver_quickstart.py

"""

from mcp.server.mcpserver import MCPServer

# Create an MCP server

mcp = MCPServer("Demo")

# Add an addition tool

@mcp.tool()

def add(a: int, b: int) -> int:

    """Add two numbers"""

    return a + b

# Add a dynamic greeting resource

@mcp.resource("greeting://{name}")

def get_greeting(name: str) -> str:

    """Get a personalized greeting"""

    return f"Hello, {n

RemediationAI

The problem is that the `mcpserver_quickstart.py` example file may contain prompt handlers that interpolate user input unsafely. If any prompt handlers are present or added to this file, they must not use f-strings to interpolate user-controlled parameters directly into prompt text. Wrap any user parameters in `<untrusted>...</untrusted>` tags or use `json.dumps()` for escaping. Verify by adding a test prompt handler with a malicious payload like `{name}'; DROP TABLE users; --` and confirming that it is rendered as literal text, not executed.

Evidence

from mcp.server.mcpserver import MCPServer

from mcp.types import (

    Completion,

    CompletionArgument,

    CompletionContext,

    PromptReference,

    ResourceTemplateReference,

)

mcp = MCPServer(name="Example")

@mcp.resource("github://repos/{owner}/{repo}")

def github_repo(owner: str, repo: str) -> str:

    """GitHub repository resource."""

    return f"Repository: {owner}/{repo}"

@mcp.prompt(description="Code review prompt")

def review_code(language: str, code: str) -> str:

    """Gen

RemediationAI

The problem is that the `completion.py` example file registers a resource handler with path parameters (`owner` and `repo`) that could be interpolated unsafely into prompt or resource text. Ensure that any prompt or resource content returned by handlers wraps user-controlled parameters in `<untrusted>...</untrusted>` tags or uses `json.dumps()` for escaping. Verify by passing a malicious `owner` or `repo` value like `<script>alert('xss')</script>` and confirming that it is rendered as literal text in the resource output, not as executable code.

Evidence

# MCP Python SDK

!!! info "You are viewing the in-development v2 documentation"

    For the current stable release, see the [v1.x documentation](https://py.sdk.modelcontextprotocol.io/).

The **Model Context Protocol (MCP)** allows applications to provide context for LLMs in a standardized way, separating the concerns of providing context from the actual LLM interaction.

This Python SDK implements the full MCP specification, making it easy to:

- **Build MCP servers** that expose resources, pr

RemediationAI

The problem is that docs/index.md documents the MCP Python SDK without explicitly warning against prompt injection vulnerabilities in prompt handlers. Add a security section to docs/index.md that warns developers to never interpolate user input directly into prompt text using f-strings or `.format()`, and instead mandate wrapping untrusted parameters in `<untrusted>...</untrusted>` tags or using `json.dumps()`. Include a vulnerable vs. safe code example. Verify by reviewing the documentation and confirming that all prompt handler examples follow the safe escaping pattern.

Evidence

from __future__ import annotations

from datetime import datetime

from typing import Annotated, Any, Final, Generic, Literal, TypeAlias, TypeVar

from pydantic import BaseModel, ConfigDict, Field, FileUrl, TypeAdapter

from pydantic.alias_generators import to_camel

from typing_extensions import NotRequired, TypedDict

from mcp.types.jsonrpc import RequestId

LATEST_PROTOCOL_VERSION = "2025-11-25"

"""The latest version of the Model Context Protocol.

You can find the latest specification at https:

RemediationAI

The problem is that `src/mcp/types/_types.py` defines MCP type structures but does not enforce authentication on HTTP routes that expose `tools/list`. The file itself is a type definition module and does not mount routes, but the issue indicates that routes using these types lack auth middleware. Add authentication middleware at the router level in the HTTP server setup (typically in a separate server file) by applying a `@require_auth` decorator or `Depends(verify_jwt)` dependency to all routes that handle MCP operations. Verify by attempting to call `GET /tools/list` without credentials — the server should return a 401 or 403 error.

Evidence

"""MCP Server Module

This module provides a framework for creating an MCP (Model Context Protocol) server.

It allows you to easily define and handle various types of requests and notifications

using constructor-based handler registration.

Usage:

1. Define handler functions:

   async def my_list_tools(ctx, params):

       return types.ListToolsResult(tools=[...])

   async def my_call_tool(ctx, params):

       return types.CallToolResult(content=[...])

2. Create a Server instance with on_* han

RemediationAI

The problem is that `src/mcp/server/lowlevel/server.py` provides a framework for registering MCP handlers but does not enforce authentication on HTTP routes that expose `tools/list`. Add authentication middleware to the HTTP transport layer by wrapping all route handlers with an auth check (e.g., `Depends(get_current_user)` in FastAPI or a custom middleware function). Verify by attempting to call `tools/list` without valid credentials — the server should reject the request with a 401 or 403 status code.

Evidence

"""Run from the repository root:

uv run examples/snippets/servers/streamable_config.py

"""

from mcp.server.mcpserver import MCPServer

mcp = MCPServer("StatelessServer")

# Add a simple tool to demonstrate the server

@mcp.tool()

def greet(name: str = "World") -> str:

    """Greet someone by name."""

    return f"Hello, {name}!"

# Run server with streamable_http transport

# Transport-specific options (stateless_http, json_response) are passed to run()

if __name__ == "__main__":

    # Stateles

RemediationAI

The problem is that `examples/snippets/servers/streamable_config.py` binds an MCP server to localhost without enforcing authentication, making it vulnerable to DNS-rebinding attacks where a malicious web page can invoke tools from the user's browser. Change the server startup to use stdio transport instead of HTTP by modifying `mcp.run(transport="stdio")` or, if HTTP is required, add authentication by implementing a token-based check (e.g., `Authorization: Bearer <token>`) on every request. Verify by starting the server and confirming that requests without a valid token are rejected with a 401 error.

Evidence

"""MCPServer quickstart example.

Run from the repository root:

    uv run examples/snippets/servers/mcpserver_quickstart.py

"""

from mcp.server.mcpserver import MCPServer

# Create an MCP server

mcp = MCPServer("Demo")

# Add an addition tool

@mcp.tool()

def add(a: int, b: int) -> int:

    """Add two numbers"""

    return a + b

# Add a dynamic greeting resource

@mcp.resource("greeting://{name}")

def get_greeting(name: str) -> str:

    """Get a personalized greeting"""

    return f"Hello, {n

RemediationAI

The problem is that `examples/snippets/servers/mcpserver_quickstart.py` binds an MCP server to localhost without enforcing authentication, making it vulnerable to DNS-rebinding attacks. Switch to stdio transport by calling `mcp.run(transport="stdio")` instead of HTTP, or add authentication middleware that validates an `Authorization` header on every request. Verify by starting the server and attempting to call a tool without authentication — the server should reject the request.

Evidence

#!/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 base64

import json

import logging

import click

from mcp.server import ServerRequestContext

from mcp.server.mcpserver import Context, MCPServer

from mcp.server.mcpserver.prompts.base import UserMessage

from mcp.server.streamable_http import EventCallback, EventMessage, EventStore

from mcp.type

RemediationAI

The problem is that `examples/servers/everything-server/mcp_everything_server/server.py` binds an MCP server to localhost without enforcing authentication, making it vulnerable to DNS-rebinding attacks. Add authentication to the HTTP transport layer by implementing a token-based check (e.g., `Authorization: Bearer <token>`) on every request, or switch to stdio transport. Verify by attempting to invoke a tool without valid credentials — the server should return a 401 error.

Evidence

"""Run from the repository root:

uv run examples/snippets/servers/oauth_server.py

"""

from pydantic import AnyHttpUrl

from mcp.server.auth.provider import AccessToken, TokenVerifier

from mcp.server.auth.settings import AuthSettings

from mcp.server.mcpserver import MCPServer

class SimpleTokenVerifier(TokenVerifier):

    """Simple token verifier for demonstration."""

    async def verify_token(self, token: str) -> AccessToken | None:

        pass  # This is where you would implement actual to

RemediationAI

The problem is that `examples/snippets/servers/oauth_server.py` binds an MCP server to localhost with OAuth authentication but does not validate the `Host` header, allowing DNS-rebinding attacks to bypass same-origin checks. Add host header validation by checking that `req.headers.host` matches an allow-list of expected hostnames (e.g., `localhost:8000`) before processing any request. Implement this as middleware that runs before the auth check. Verify by making a request with a spoofed `Host` header — the server should reject it with a 400 error.

LLM consensus

Evidence

# MCP Python SDK

<div align="center">

<strong>Python implementation of the Model Context Protocol (MCP)</strong>

[![PyPI][pypi-badge]][pypi-url]

[![MIT licensed][mit-badge]][mit-url]

[![Python Version][python-badge]][python-url]

[![Documentation][docs-badge]][docs-url]

[![Protocol][protocol-badge]][protocol-url]

[![Specification][spec-badge]][spec-url]

</div>

<!-- TODO(v2): Move this content back to README.md when v2 is released -->

> [!IMPORTANT]

> **This documents v2 of the SDK (currentl

RemediationAI

The problem is that README.v2.md documents MCP server setup without mentioning host header validation, which could lead developers to deploy servers vulnerable to DNS-rebinding attacks even with authentication in place. Add a security section to README.v2.md that warns developers to validate the `Host` header against an allow-list of expected hostnames when binding to localhost. Include a code example showing how to implement host header validation middleware. Verify by reviewing the documentation and confirming that all HTTP server examples include host validation.

Evidence

# MCP Python SDK

<div align="center">

<strong>Python implementation of the Model Context Protocol (MCP)</strong>

[![PyPI][pypi-badge]][pypi-url]

[![MIT licensed][mit-badge]][mit-url]

[![Python Version][python-badge]][python-url]

[![Documentation][docs-badge]][docs-url]

[![Protocol][protocol-badge]][protocol-url]

[![Specification][spec-badge]][spec-url]

</div>

<!-- TODO(v2): Replace this README with README.v2.md when v2 is released -->

> [!NOTE]

> **This README documents v1.x of the MCP Pyt

RemediationAI

The problem is that README.md documents MCP server setup without mentioning host header validation, which could lead developers to deploy servers vulnerable to DNS-rebinding attacks even with authentication in place. Add a security section to README.md that warns developers to validate the `Host` header against an allow-list of expected hostnames when binding to localhost. Include a code example showing how to implement host header validation middleware. Verify by reviewing the documentation and confirming that all HTTP server examples include host validation.

Evidence

"""MCPServer quickstart example.

Run from the repository root:

    uv run examples/snippets/servers/mcpserver_quickstart.py

"""

from mcp.server.mcpserver import MCPServer

# Create an MCP server

mcp = MCPServer("Demo")

# Add an addition tool

@mcp.tool()

def add(a: int, b: int) -> int:

    """Add two numbers"""

    return a + b

# Add a dynamic greeting resource

@mcp.resource("greeting://{name}")

def get_greeting(name: str) -> str:

    """Get a personalized greeting"""

    return f"Hello, {n

RemediationAI

The problem is that `examples/snippets/servers/mcpserver_quickstart.py` binds an MCP server to localhost without validating the `Host` header, making it vulnerable to DNS-rebinding attacks. Add host header validation middleware that checks `req.headers.host` against an allow-list (e.g., `["localhost:8000", "127.0.0.1:8000"]`) before processing any request. Verify by making a request with a spoofed `Host` header like `evil.com` — the server should reject it with a 400 error.

Evidence

"""Run from the repository root:

uv run examples/snippets/servers/streamable_config.py

"""

from mcp.server.mcpserver import MCPServer

mcp = MCPServer("StatelessServer")

# Add a simple tool to demonstrate the server

@mcp.tool()

def greet(name: str = "World") -> str:

    """Greet someone by name."""

    return f"Hello, {name}!"

# Run server with streamable_http transport

# Transport-specific options (stateless_http, json_response) are passed to run()

if __name__ == "__main__":

    # Stateles

RemediationAI

The problem is that `examples/snippets/servers/streamable_config.py` binds an MCP server to localhost without validating the `Host` header, making it vulnerable to DNS-rebinding attacks. Add host header validation middleware that checks `req.headers.host` against an allow-list of expected hostnames before processing any request. Verify by attempting to access the server with a spoofed `Host` header — the request should be rejected.

Evidence

#!/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 base64

import json

import logging

import click

from mcp.server import ServerRequestContext

from mcp.server.mcpserver import Context, MCPServer

from mcp.server.mcpserver.prompts.base import UserMessage

from mcp.server.streamable_http import EventCallback, EventMessage, EventStore

from mcp.type

RemediationAI

The problem is that `examples/servers/everything-server/mcp_everything_server/server.py` binds an MCP server to localhost without validating the `Host` header, making it vulnerable to DNS-rebinding attacks. Add host header validation middleware that checks `req.headers.host` against an allow-list of expected hostnames (e.g., `localhost`, `127.0.0.1`) before processing any request. Verify by making a request with a spoofed `Host` header — the server should reject it with a 400 error.

Evidence

# Migration Guide: v1 to v2

This guide covers the breaking changes introduced in v2 of the MCP Python SDK and how to update your code.

## Overview

Version 2 of the MCP Python SDK introduces several breaking changes to improve the API, align with the MCP specification, and provide better type safety.

## Breaking Changes

### `streamablehttp_client` removed

The deprecated `streamablehttp_client` function has been removed. Use `streamable_http_client` instead.

**Before (v1):**

```python

from

RemediationAI

The problem is that `docs/migration.md` documents MCP server setup without mentioning host header validation, which could lead developers to deploy servers vulnerable to DNS-rebinding attacks. Add a security section to docs/migration.md that warns developers to validate the `Host` header against an allow-list of expected hostnames when binding to localhost. Include a code example showing how to implement host header validation. Verify by reviewing the documentation and confirming that all migration examples include host validation.

Evidence

# MCP Python SDK

!!! info "You are viewing the in-development v2 documentation"

    For the current stable release, see the [v1.x documentation](https://py.sdk.modelcontextprotocol.io/).

The **Model Context Protocol (MCP)** allows applications to provide context for LLMs in a standardized way, separating the concerns of providing context from the actual LLM interaction.

This Python SDK implements the full MCP specification, making it easy to:

- **Build MCP servers** that expose resources, pr

RemediationAI

The problem is that `docs/index.md` documents the MCP Python SDK without mentioning host header validation, which could lead developers to deploy servers vulnerable to DNS-rebinding attacks. Add a security section to docs/index.md that warns developers to validate the `Host` header against an allow-list of expected hostnames when binding to localhost. Include a code example showing how to implement host header validation middleware. Verify by reviewing the documentation and confirming that all server examples include host validation.

Evidence

# Or for SSE

mcp = MCPServer("Server")

mcp.run(transport="sse", host="0.0.0.0", port=9000, sse_path="/events")

```

**For mounted apps:**

RemediationAI

The problem is that `docs/migration.md` shows an example of binding an MCP server to `0.0.0.0` (all network interfaces), which exposes the server to the entire network. Change the example to bind to `127.0.0.1` instead by modifying `mcp.run(transport="sse", host="127.0.0.1", port=9000, sse_path="/events")`. Verify by starting the server and confirming that it only listens on the loopback interface: `netstat -an | grep 9000` should show `127.0.0.1:9000`, not `0.0.0.0:9000`.

Evidence

mcp.run(transport="streamable-http")

# Or for SSE

mcp = FastMCP("Server", host="0.0.0.0", port=9000, sse_path="/events")

mcp.run(transport="sse")

```

RemediationAI

The problem is that `docs/migration.md` shows an example of binding an MCP server to `0.0.0.0` (all network interfaces), which exposes the server to the entire network. Change the example to bind to `127.0.0.1` instead by modifying `FastMCP("Server", host="127.0.0.1", port=9000, sse_path="/events")`. Verify by starting the server and confirming that it only listens on the loopback interface: `netstat -an | grep 9000` should show `127.0.0.1:9000`, not `0.0.0.0:9000`.

Evidence

message = params.message

    if not url:

        print("Error: No URL provided in elicitation request")

        return types.ElicitResult(action="cancel")

    # Reject dangerous URL schemes before prompting the user

RemediationAI

The problem is that `examples/snippets/clients/url_elicitation_client.py` prints user-controlled messages to the terminal without sanitizing ANSI escape sequences, allowing an attacker to inject cursor-control codes that rewrite output or hide commands. Replace the `print(f"Error: No URL provided in elicitation request")` call with a sanitized version by stripping ANSI escape sequences from `params.message` before printing, using a library like `re.sub(r'\x1b\[[0-9;]*m', '', params.message)` or `bleach.clean()`. Verify by passing a message containing ANSI codes like `\x1b[2J` (clear screen) and confirming that the terminal is not cleared.

Evidence

async def handle_progress(ctx: ServerRequestContext, params: ProgressNotificationParams) -> None:

    print(f"Progress: {params.progress}/{params.total}")

server = Server("my-server", on_progress=handle_progress)

```

RemediationAI

The problem is that `docs/migration.md` shows an example that prints user-controlled progress values to the terminal without sanitizing ANSI escape sequences. Replace `print(f"Progress: {params.progress}/{params.total}")` with a sanitized version by stripping ANSI codes from both `params.progress` and `params.total` before printing, using `re.sub(r'\x1b\[[0-9;]*m', '', str(params.progress))`. Verify by passing progress values containing ANSI codes and confirming that the terminal output is not manipulated.

Evidence

async def handle_elicitation(context, params: ElicitRequestParams) -> ElicitResult:

    # Display the message to the user

    print(f"Server asks: {params.message}")

    # Collect user input (this is a simplified example)

    response = input("Your response (y/n): ")

RemediationAI

The problem is that `docs/experimental/tasks-client.md` shows an example that prints user-controlled elicitation messages to the terminal without sanitizing ANSI escape sequences. Replace `print(f"Server asks: {params.message}")` with a sanitized version by stripping ANSI codes from `params.message` before printing, using `re.sub(r'\x1b\[[0-9;]*m', '', params.message)`. Verify by passing a message containing ANSI codes and confirming that the terminal is not manipulated.

Evidence

async def handle_sampling_message(

    context: ClientRequestContext, params: types.CreateMessageRequestParams

) -> types.CreateMessageResult:

    print(f"Sampling request: {params.messages}")

    return types.CreateMessageResult(

        role="assistant",

        content=types.TextContent(

RemediationAI

The problem is that `README.v2.md` shows an example that prints user-controlled sampling request messages to the terminal without sanitizing ANSI escape sequences. Replace `print(f"Sampling request: {params.messages}")` with a sanitized version by stripping ANSI codes from `params.messages` before printing, using `re.sub(r'\x1b\[[0-9;]*m', '', str(params.messages))`. Verify by passing a message containing ANSI codes and confirming that the terminal output is not manipulated.

Evidence

async def handle_sampling_message(

    context: RequestContext[ClientSession, None], params: types.CreateMessageRequestParams

) -> types.CreateMessageResult:

    print(f"Sampling request: {params.messages}")

    return types.CreateMessageResult(

        role="assistant",

        content=types.TextContent(

RemediationAI

The problem is that `README.md` shows an example that prints user-controlled sampling request messages to the terminal without sanitizing ANSI escape sequences. Replace `print(f"Sampling request: {params.messages}")` with a sanitized version by stripping ANSI codes from `params.messages` before printing, using `re.sub(r'\x1b\[[0-9;]*m', '', str(params.messages))`. Verify by passing a message containing ANSI codes and confirming that the terminal output is not manipulated.

Evidence

async def handle_sampling_message(

    context: ClientRequestContext, params: types.CreateMessageRequestParams

) -> types.CreateMessageResult:

    print(f"Sampling request: {params.messages}")

    return types.CreateMessageResult(

        role="assistant",

        content=types.TextContent(

RemediationAI

The problem is that `examples/snippets/clients/stdio_client.py` prints user-controlled sampling request messages to the terminal without sanitizing ANSI escape sequences. Replace `print(f"Sampling request: {params.messages}")` with a sanitized version by stripping ANSI codes from `params.messages` before printing, using `re.sub(r'\x1b\[[0-9;]*m', '', str(params.messages))`. Verify by passing a message containing ANSI codes and confirming that the terminal is not manipulated.

Evidence

async def elicitation_callback(context, params: ElicitRequestParams) -> ElicitResult:

    print(f"\n[Elicitation] {params.message}")

    response = input("Confirm? (y/n): ")

    return ElicitResult(action="accept", content={"confirm": response.lower() == "y"})

RemediationAI

The problem is that `docs/experimental/tasks-client.md` shows an example that prints user-controlled elicitation messages to the terminal without sanitizing ANSI escape sequences. Replace `print(f"\n[Elicitation] {params.message}")` with a sanitized version by stripping ANSI codes from `params.message` before printing, using `re.sub(r'\x1b\[[0-9;]*m', '', params.message)`. Verify by passing a message containing ANSI codes and confirming that the terminal is not manipulated.

Evidence

def main() -> None:

    """Main entry point for the conformance client."""

    if len(sys.argv) < 2:

        print(f"Usage: {sys.argv[0]} <server-url>", file=sys.stderr)

        sys.exit(1)

    server_url = sys.argv[1]

RemediationAI

The problem is that `.github/actions/conformance/client.py` prints user-controlled server URLs to stderr without sanitizing ANSI escape sequences. Replace `print(f"Usage: {sys.argv[0]} <server-url>", file=sys.stderr)` with a sanitized version by stripping ANSI codes from `sys.argv[0]` before printing, using `re.sub(r'\x1b\[[0-9;]*m', '', sys.argv[0])`. Verify by passing a URL containing ANSI codes and confirming that the terminal output is not manipulated.

Evidence

params: ElicitRequestParams,

) -> ElicitResult:

    """Handle elicitation requests from the server."""

    print(f"\n[Elicitation] Server asks: {params.message}")

    # Simple terminal prompt

    response = input("Your response (y/n): ").strip().lower()

RemediationAI

The problem is that `examples/clients/simple-task-interactive-client/mcp_simple_task_interactive_client/main.py` prints user-controlled elicitation messages to the terminal without sanitizing ANSI escape sequences. Replace `print(f"\n[Elicitation] Server asks: {params.message}")` with a sanitized version by stripping ANSI codes from `params.message` before printing, using `re.sub(r'\x1b\[[0-9;]*m', '', params.message)`. Verify by passing a message containing ANSI codes and confirming that the terminal is not manipulated.

Evidence

return f"Elicitation result: action={result.action}, content={content}"

    except Exception as e:

        return f"Elicitation not supported or error: {str(e)}"

class EnumSchemasTestSchema(BaseModel):

RemediationAI

The problem is that `examples/servers/everything-server/mcp_everything_server/server.py` returns full exception details to the caller via `str(e)`, which leaks internal paths, library versions, and implementation details useful for reconnaissance. Replace `return f"Elicitation not supported or error: {str(e)}"` with a generic error message like `return "Elicitation not supported or error: An error occurred"` and log the full exception internally using `logging.exception("Elicitation error")`. Verify by triggering an exception and confirming that the response contains only a generic message, not the full traceback.

Evidence

return f"LLM response: {model_response}"

    except Exception as e:

        return f"Sampling not supported or error: {str(e)}"

class UserResponse(BaseModel):

RemediationAI

The problem is that `examples/servers/everything-server/mcp_everything_server/server.py` returns full exception details to the caller via `str(e)`, which leaks internal paths, library versions, and implementation details. Replace `return f"Sampling not supported or error: {str(e)}"` with a generic error message like `return "Sampling not supported or error: An error occurred"` and log the full exception internally using `logging.exception("Sampling error")`. Verify by triggering an exception and confirming that the response contains only a generic message.

Evidence

return f"User response: action={result.action}, content={content}"

    except Exception as e:

        return f"Elicitation not supported or error: {str(e)}"

class SEP1034DefaultsSchema(BaseModel):

RemediationAI

The problem is that `examples/servers/everything-server/mcp_everything_server/server.py` returns full exception details to the caller via `str(e)`, which leaks internal paths, library versions, and implementation details. Replace `return f"Elicitation not supported or error: {str(e)}"` with a generic error message like `return "Elicitation not supported or error: An error occurred"` and log the full exception internally using `logging.exception("Elicitation error")`. Verify by triggering an exception and confirming that the response contains only a generic message.

Evidence

return f"Elicitation completed: action={result.action}, content={content}"

    except Exception as e:

        return f"Elicitation not supported or error: {str(e)}"

@mcp.tool()

RemediationAI

The problem is that `examples/servers/everything-server/mcp_everything_server/server.py` returns full exception details to the caller via `str(e)`, which leaks internal paths, library versions, and implementation details. Replace `return f"Elicitation not supported or error: {str(e)}"` with a generic error message like `return "Elicitation not supported or error: An error occurred"` and log the full exception internally using `logging.exception("Elicitation error")`. Verify by triggering an exception and confirming that the response contains only a generic message.

Evidence

except MCPError:

            raise

        except Exception as e:

            return CallToolResult(content=[TextContent(type="text", text=str(e))], is_error=True)

        if isinstance(result, CallToolResult):

            return result

        if isinstance(result, tuple) and len(result) == 2:

RemediationAI

The problem is that `src/mcp/server/mcpserver/server.py` returns full exception details to the caller via `str(e)` in the `CallToolResult`, which leaks internal paths, library versions, and implementation details. Replace `return CallToolResult(content=[TextContent(type="text", text=str(e))], is_error=True)` with a generic error message like `return CallToolResult(content=[TextContent(type="text", text="Tool execution failed")], is_error=True)` and log the full exception internally using `logging.exception("Tool execution error")`. Verify by triggering a tool exception and confirming that the response contains only a generic message.

Evidence

# Try both npx.cmd and npx.exe on Windows

        for cmd in ["npx.cmd", "npx.exe", "npx"]:

            try:

                subprocess.run([cmd, "--version"], check=True, capture_output=True, shell=True)

                return cmd

            except subprocess.CalledProcessError:

                continue

RemediationAI

The problem is that `src/mcp/cli/cli.py` calls `subprocess.run()` without specifying a timeout, allowing a hung or malicious subprocess to block indefinitely and exhaust system resources. Add a `timeout` parameter to the `subprocess.run()` call: `subprocess.run([cmd, "--version"], check=True, capture_output=True, shell=False, timeout=5)` (adjust the timeout value as appropriate). Verify by running the CLI and confirming that it completes within the timeout; then test with a command that hangs to confirm that a `subprocess.TimeoutExpired` exception is raised after the timeout expires.

Evidence

async def update_importance(user_embedding: list[float], deps: Deps):

    async with deps.pool.acquire() as conn:

        rows = await conn.fetch("SELECT id, importance, access_count, embedding FROM memories")

        for row in rows:

            memory_embedding = row["embedding"]

            similarity = cosine_similarity(user_embedding, memory_embedding)

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

LLM consensus

Evidence

# /// script

# dependencies = ["pydantic-ai-slim[openai]", "asyncpg", "numpy", "pgvector"]

# ///

# uv pip install 'pydantic-ai-slim[openai]' asyncpg numpy pgvector

"""Recursive memory system inspired by the human brain's clustering of memories.

Uses OpenAI's 'text-embedding-3-small' model and pgvector for efficient

similarity search.

"""

import asyncio

import math

import os

from dataclasses import dataclass

from datetime import datetime, timezone

from pathlib import Path

from typing import An

Remediation

Pick one confirmation mechanism and add it to the destructive handler: - make the client confirm via `elicit()` before touching the sink, - accept a `dry_run` / `dryRun` boolean input and branch off the sink when true (return what would change), - require an `idempotency_key` / `idempotencyKey` the caller provides on the first call and echoes back to commit, - require a `confirmation` / `confirm` token as an input field. If the destructive call is genuinely irrelevant to the tool

LLM consensus

Evidence

description="Query the database",

            inputSchema={

                "type": "object",

                "properties": {"query": {"type": "string", "description": "SQL query to execute"}},

                "required": ["query"],

            },

        )

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

# see OAuthClientMetadata; we only support `code`

    response_type: Literal["code"] = Field(..., description="Must be 'code' for authorization code flow")

    code_challenge: str = Field(..., description="PKCE code challenge")

    code_challenge_method: Literal["S256"] = Field("S256", description="PKCE code challenge method, must be S256")

    state: str | None = Field(None, description="Optional state parameter")

    scope: str | None = Field(

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

"""Parameters for initializing an sse_client."""

    # The endpoint URL.

    url: str

    # Optional headers to include in requests.

    headers: dict[str, Any] | None = None

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

"""Parameters for initializing a streamable_http_client."""

    # The endpoint URL.

    url: str

    # Optional headers to include in requests.

    headers: dict[str, Any] | None = None

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

description="Query the database",

                input_schema={

                    "type": "object",

                    "properties": {"query": {"type": "string", "description": "SQL query to execute"}},

                    "required": ["query"],

                },

            )

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

"type": "object",

                    "required": ["url"],

                    "properties": {

                        "url": {

                            "type": "string",

                            "description": "URL to fetch",

                        }

                    },

                },

            )

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

class HttpResource(Resource):

    """A resource that reads from an HTTP endpoint."""

    url: str = Field(description="URL to fetch content from")

    mime_type: str = Field(default="application/json", description="MIME type of the resource content")

    async def read(self) -> str | bytes:

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

class AuthorizationParams(BaseModel):

    state: str | None

    scopes: list[str] | None

    code_challenge: str

    redirect_uri: AnyUrl

    redirect_uri_provided_explicitly: bool

    resource: str | None = None  # RFC 8707 resource indicator

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

message: str

    """The message to present to the user explaining why the interaction is needed."""

    url: str

    """The URL that the user should navigate to."""

    elicitation_id: str

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

class AuthorizationCode(BaseModel):

    code: str

    scopes: list[str]

    expires_at: float

    client_id: str

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

scopes: list[str]

    expires_at: float

    client_id: str

    code_challenge: str

    redirect_uri: AnyUrl

    redirect_uri_provided_explicitly: bool

    resource: str | None = None  # RFC 8707 resource indicator

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

# we use the client_secret param, per https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1

    client_secret: str | None = None

    # See https://datatracker.ietf.org/doc/html/rfc7636#section-4.5

    code_verifier: str = Field(..., description="PKCE code verifier")

    # RFC 8707 resource indicator

    resource: str | None = Field(None, description="Resource indicator for the token")

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

class StdioServerParameters(BaseModel):

    command: str

    """The executable to run to start the server."""

    args: list[str] = Field(default_factory=list)

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

class AuthorizationCodeRequest(BaseModel):

    # See https://datatracker.ietf.org/doc/html/rfc6749#section-4.1.3

    grant_type: Literal["authorization_code"]

    code: str = Field(..., description="The authorization code")

    redirect_uri: AnyUrl | None = Field(None, description="Must be the same as redirect URI provided in /authorize")

    client_id: str

    # we use the client_secret param, per https://datatracker.ietf.org/doc/html/rfc6749#section-2.3.1

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

description="Query the database",

                input_schema={

                    "type": "object",

                    "properties": {"query": {"type": "string", "description": "SQL query to execute"}},

                    "required": ["query"],

                },

            )

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

{

  "mcpServers": {

    "sqlite": {

      "command": "uvx",

      "args": ["mcp-server-sqlite", "--db-path", "./test.db"]

    },

    "puppeteer": {

      "command": "npx",

      "args": ["-y", "@modelcontextprotocol/server-puppeteer"]

    }

  }

}

Remediation

Declare a real authentication mechanism in the manifest, matching what the running server actually enforces: - `"auth": "bearer"` with a token scheme documented for callers - `"auth": "oauth"` / `"oauth2": { ... }` for delegated flows - `"apiKey": { "header": "X-API-Key", "prefix": "..." }` - `"mtls": true` when client certificates are required If the server is intentionally unauthenticated (stdio-only, local developer tool, trusted-host network), document the assumption in the manifest via a `"

Evidence

"""URL Elicitation Client Example.

Demonstrates how clients handle URL elicitation requests from servers.

This is the Python equivalent of TypeScript SDK's elicitationUrlExample.ts,

focused on URL elicitation patterns without OAuth complexity.

Features demonstrated:

1. Client elicitation capability declaration

2. Handling elicitation requests from servers via callback

3. Catching UrlElicitationRequiredError from tool calls

4. Browser interaction with security warnings

5. Interactive CLI for te

Remediation

Tool descriptions should describe what the tool does — not what the model should do with other tools. If a tool's correct operation legitimately requires another tool to be called, document that as a `composition` requirement in human-readable docs and let the calling code orchestrate, not the LLM. If the directive phrasing is coming from external content the tool retrieved (RAG, web fetch), wrap in `<untrusted>` tags and rely on the system prompt to flag tag-bound content as data, not instruct

Evidence

from __future__ import annotations

from datetime import datetime

from typing import Annotated, Any, Final, Generic, Literal, TypeAlias, TypeVar

from pydantic import BaseModel, ConfigDict, Field, FileUrl, TypeAdapter

from pydantic.alias_generators import to_camel

from typing_extensions import NotRequired, TypedDict

from mcp.types.jsonrpc import RequestId

LATEST_PROTOCOL_VERSION = "2025-11-25"

"""The latest version of the Model Context Protocol.

You can find the latest specification at https:

Remediation

Tool descriptions should describe what the tool does — not what the model should do with other tools. If a tool's correct operation legitimately requires another tool to be called, document that as a `composition` requirement in human-readable docs and let the calling code orchestrate, not the LLM. If the directive phrasing is coming from external content the tool retrieved (RAG, web fetch), wrap in `<untrusted>` tags and rely on the system prompt to flag tag-bound content as data, not instruct

Evidence