ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-21 21:14:17 +02:00
Code Issues Projects Releases Packages Wiki Activity

Respell server protocol restriction as a version allowlist

Browse source 
Replace `min_protocol_version` with `protocol_versions`, an allowlist of MCP
protocol versions taking the SDK's own era tuples. Versions are an enumerated
set, not an ordered scalar, so enforcement is set membership rather than a
bound -- which is what makes handshake-only expressible.

Generalize the startup coherence check into a capability -> required-versions
map, and make it silent unless a version set was declared.
This commit is contained in:
Jeremiah Lowin 2026-07-20 12:20:56 -04:00
commit f5fca97c81
No known key found for this signature in database
9 changed files with 931 additions and 581 deletions
Download patch file Download diff file Expand all files Collapse all files

12
docs/development/v4-notes/change-register.mdx
View file

@ -380,13 +380,17 @@ A tool can gather client input across rounds on a `2026-07-28` call by returning
*Verify:* `fastmcp_slim/fastmcp/server/context.py` (`input_responses`/`request_state` properties), `fastmcp_slim/fastmcp/server/low_level.py` (`RequestStateBoundary` install), `fastmcp_slim/fastmcp/server/server.py` (`request_state_security` param), `fastmcp_slim/fastmcp/server/mixins/mcp_operations.py` (`_on_call_tool` input-required passthrough + era gate), `fastmcp_slim/fastmcp/tools/base.py` (`InputRequiredToolResult`), `tests/server/test_mrtr_guards.py`.
### Server protocol floor — New (opt-in feature, provisional API)
### Server protocol-version restriction — New (opt-in feature)
A server can declare the minimum MCP protocol version it requires, so a client that cannot meet it is refused at connection time instead of failing mid tool-call. The motivating case is modern-only guard tools (a tool returning `InputRequiredResult`, SEP-2322): under `FastMCP(min_protocol_version="2026-07-28")` a legacy client is refused during the initialize handshake with a clear `-32602` error naming the required version and pointing at the modern protocol, rather than discovering the incompatibility inside a call. Enforcement runs through FastMCP's entry in the SDK's middleware layer at the two negotiation points the framework owns: the initialize handshake (the negotiated handshake version is compared against the floor before the handshake commits) and `server/discover` (the modern era is the newest era, so it satisfies any currently declarable floor — discovery is never refused by a floor today). At startup a conservative, warning-only coherence check flags declared/registered conflicts: guard tools under a non-modern floor (handshake clients would fail mid-call) and a modern floor paired with a `"fallback"` sampling handler (the modern era forbids the back-channel, so the fallback is dead). Runtime-only back-channel usage (`ctx.elicit`/`ctx.sample`/`ctx.list_roots`) has no reliable static signal and is not inferred. The default is no floor — every era is served, fully backward compatible.
A server can declare the set of MCP protocol versions it serves, so a client that shares none of them is refused at connection time instead of failing mid tool-call. The declaration is an allowlist expressed in the SDK's own units — `FastMCP(protocol_versions=MODERN_PROTOCOL_VERSIONS)`, `FastMCP(protocol_versions=HANDSHAKE_PROTOCOL_VERSIONS)`, or exact strings like `["2026-07-28"]` — so authors pass the SDK's era tuples rather than FastMCP-invented alias vocabulary, and those tuples grow on their own when the SDK adds a revision to an era. An unrecognized version string is a `ValueError` at construction.
The public spelling (`min_protocol_version=`) is **provisional** and expected to change (the min-without-max shape is under review); the enforcement hook and inference rules are stable regardless of the eventual keyword. No `settings.py` entry yet.
Enforcement is **set membership, not a minimum**, matching the SDK's stated model that versions are an enumerated set rather than an ordered scalar (`mcp_types/version.py`). That is what makes handshake-only expressible: the modern era is numerically newer but *removed* the server-initiated back-channel (`ctx.elicit`/`ctx.sample`/`ctx.list_roots`), so a server built on the back-channel needs the handshake era specifically — a minimum-version model could not say that. Both connection paths are covered: the initialize handshake is refused before it commits, and modern connections are checked on the request itself, since a client pinned to a modern version never probes `server/discover`. Refusals use the spec-standard `-32022` unsupported-protocol-version error with the server's `supported` list, so a `mode="auto"` client refused at discovery by a handshake-only server reads the handshake versions out of the error and completes over `initialize` on its own.
*Verify:* `fastmcp_slim/fastmcp/server/protocol_floor.py` (validation, negotiation mirror, enforcement, coherence check), `fastmcp_slim/fastmcp/server/low_level.py` (`enforce_handshake_floor` in the initialize path), `fastmcp_slim/fastmcp/server/mixins/lifespan.py` (startup coherence call), `tests/server/test_protocol_floor.py`.
The startup coherence check is now a **capability map** rather than ad-hoc cases: each entry names a capability, the protocol versions that carry it, and a detector. Present entries are multi-round-trip guard tools (modern versions, detected via `_contains_input_required` over tool return annotations) and the client back-channel (handshake versions, detected only from the configuration-level `sampling_handler` + `"fallback"` contradiction — the runtime calls have no static signal). A future capability, notably the 2026 tasks extension, is one entry rather than a new special case. Warnings only, never a hard error, and **silent unless a version set was declared** — a server with an ordinary guard tool and no declaration is fine, and warning there would train people to ignore warnings.
The default is `None`: every protocol version the SDK supports is served. Flipping a server default would disconnect existing clients rather than degrade, so restriction stays strictly opt-in. No `settings.py` entry — this is per-server configuration, not global.
*Verify:* `fastmcp_slim/fastmcp/server/protocol_versions.py` (validation, negotiation mirror, enforcement, capability map, coherence check), `fastmcp_slim/fastmcp/server/low_level.py` (handshake veto in the initialize path, per-request check in the root dispatch), `fastmcp_slim/fastmcp/server/mixins/lifespan.py` (startup coherence call), `tests/server/test_protocol_versions.py`.
### The xfail register — Known gap

147
docs/development/v4-notes/protocol-2026.mdx
View file

@ -50,71 +50,130 @@ The complete picture of what a FastMCP v4 server and client provide on the `2026
| **Pagination** | Declarative `FastMCP(list_page_size=...)` paginates all list operations in the high-level server; the client auto-paginates with cycle detection. |
| **Telemetry** | OpenTelemetry spans on by default (no-op without an exporter), SDK-aligned attributes (`mcp.method.name`, `mcp.protocol.version`, `gen_ai.*`), plus auth and provider-delegation spans; `FASTMCP_ENABLE_TELEMETRY=false` disables cleanly. |
## Server protocol floor (DRAFT — provisional API)
<Warning>
This section is a **draft** for an in-progress feature. The public spelling shown
here (`min_protocol_version=`) is **provisional** and expected to change before
release — the min-without-max shape is under review. The underlying mechanics
(connect-time enforcement and startup coherence checks) are stable regardless of
the final spelling.
</Warning>
## Restricting the protocol versions a server serves
<VersionBadge version="4.0.0" />
A server can depend on features that exist on only one protocol era. A tool that
returns an `InputRequiredResult` (the modern [guard pattern](/servers/elicitation#elicitation-on-the-modern-protocol))
works only on a `2026-07-28` connection. Today a legacy client that reaches such
a tool over the initialize handshake gets a confusing era error *mid tool-call*,
long after connecting — the failure surfaces far from its cause.
Most servers should skip this section. By default a FastMCP server serves every
protocol version the SDK supports, and that is the right setting for almost
everything — one server, every client, no configuration.
The protocol floor moves that failure to connect time. A server declares the
minimum protocol version it requires, and FastMCP refuses any handshake below it
with a clear error naming the required version:
Some servers cannot. A server can depend on a feature that exists on only one
protocol era, and the two eras are not a ladder: the modern era added the
multi-round-trip [guard pattern](/servers/elicitation#elicitation-on-the-modern-protocol)
and *removed* the server-initiated back-channel (`ctx.elicit`, `ctx.sample`,
`ctx.list_roots`) that the handshake eras provide. Neither era is a superset of
the other, so a server can legitimately need either one.
When a server's whole purpose depends on an era, a client that cannot speak it
should be told at connect time. Today it finds out mid-call: a handshake-era
client happily connects to a guard-tool server, lists its tools, calls one, and
only then gets an era error — a failure surfacing far from its cause. Declaring
the versions the server serves moves that failure to the connection.
### Declaring versions
Pass the protocol versions the server is willing to serve. The MCP SDK exposes
each era as a tuple, and those tuples are the durable way to name an era — they
grow on their own when the SDK adds a revision, so a server pinned to
`MODERN_PROTOCOL_VERSIONS` keeps working across SDK upgrades:
```python
from fastmcp import FastMCP
from fastmcp.types import InputRequiredResult
from mcp_types.version import MODERN_PROTOCOL_VERSIONS
mcp = FastMCP("guarded", min_protocol_version="2026-07-28")
mcp = FastMCP("guarded", protocol_versions=MODERN_PROTOCOL_VERSIONS)
@mcp.tool
def confirm(action: str) -> str | InputRequiredResult:
... # a modern-only guard tool
... # a guard tool: multi-round trips exist only on the modern era
```
A legacy client connecting to this server is refused during `initialize` with a
message pointing it at the modern protocol, rather than discovering the
incompatibility inside a tool call. Modern (`server/discover`) clients connect
normally — the modern era satisfies any currently declarable floor, because it is
the newest era.
A server built around the back-channel declares the other era:
**Enforcement points.** The floor is applied at the two negotiation points
FastMCP owns, both through the framework's entry in the SDK's middleware layer:
```python
from fastmcp import Context, FastMCP
from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS
- **Initialize handshake** — the connection's negotiated handshake version is
compared against the floor before the handshake commits; a version below the
floor is refused with `-32602` (invalid params).
- **`server/discover` (modern)** — the modern era is the newest era, so every
modern connection satisfies any floor a server can declare today; discovery is
therefore never refused by a floor in the current version set.
mcp = FastMCP("interviewer", protocol_versions=HANDSHAKE_PROTOCOL_VERSIONS)
**Startup coherence check.** FastMCP knows what is registered, so at startup it
warns (never fails) when the declared floor and the registered features conflict:
- A guard tool (returns `InputRequiredResult`) under a non-modern floor will fail
for handshake-era clients mid-call — the warning names the tool and recommends
a modern floor.
- A modern floor combined with a `sampling_handler` set to `"fallback"` is a dead
preference: the modern era forbids the server-initiated back-channel, so the
local handler always runs. The warning suggests `"always"` or a lower floor.
@mcp.tool
async def interview(ctx: Context) -> str:
answer = await ctx.elicit("What is your name?", response_type=str)
return f"Hello, {answer.data}"
```
Runtime-only back-channel usage (`ctx.elicit`, `ctx.sample`, `ctx.list_roots`)
has no reliable static signal and is deliberately not inferred.
You can also pin exact versions, which is what a server certified against a
single revision wants:
The default is no floor: a server without `min_protocol_version` serves every
protocol era, fully backward compatible.
```python
mcp = FastMCP("pinned", protocol_versions=["2026-07-28"])
```
Any string the SDK does not recognize raises `ValueError` at construction. A
version the SDK could never negotiate is a bug in the server, not a runtime
condition worth warning about.
### Membership, not a minimum
The declaration is a **set**, not a floor. Versions are an enumerated set rather
than an ordered scale — the SDK says so directly, and future revision
identifiers are not guaranteed to be date-shaped or sortable. A connection is
accepted when its negotiated version is a member of the declared set, which is
what makes "handshake only" expressible at all: under a minimum-version model
there is no way to say "the session era", because the modern era is numerically
newer while lacking the features that era depends on.
### What clients see
A refused connection gets the spec-standard `-32022` unsupported-protocol-version
error carrying the server's supported list, so a negotiating client treats it as
guidance rather than a dead end:
| Server declares | `mode="auto"` client | `mode="legacy"` client | Client pinned to `2026-07-28` |
| --- | --- | --- | --- |
| nothing (default) | modern | handshake | modern |
| `MODERN_PROTOCOL_VERSIONS` | modern | refused | modern |
| `HANDSHAKE_PROTOCOL_VERSIONS` | falls back to handshake | handshake | refused |
An `auto` client refused at `server/discover` by a handshake-only server reads
the handshake versions out of the error and completes the connection over
`initialize` on its own — the restriction steers negotiation instead of breaking
it. Only a client with no mutual version is refused outright, which is the point.
Enforcement covers both connection paths FastMCP owns. The initialize handshake
is refused before it commits. Modern connections are checked on the request
itself, because a client pinned to a modern version never probes
`server/discover` — refusing discovery alone would let it straight through.
<Note>
`fastmcp.Client` connects in-memory over the handshake by default, so testing a
modern-only server in-process needs `Client(mcp, mode="auto")`. Over HTTP and
stdio the default `auto` negotiation applies and no change is needed.
</Note>
### Startup coherence check
FastMCP knows what is registered, so at startup it warns — never fails — when a
declared version set cannot carry a capability the server actually uses. A
handshake-only server registering a guard tool gets a warning naming the tool; a
modern-only server configuring a `sampling_handler` with
`sampling_handler_behavior="fallback"` gets one too, because the modern era has
no back-channel for the fallback to reach.
The check is **silent unless you declared something**. A server with ten
ordinary tools and one guard tool is fine as-is: the guard tool raises a clear
era error if an old client reaches for it, and warning about that at startup
would only teach people to ignore warnings. The check fires when you asserted
something and the registration contradicts the assertion.
Detection is deliberately conservative. `ctx.elicit`, `ctx.sample`, and
`ctx.list_roots` are runtime calls with no reliable static signal, so only
configuration-level contradictions are caught — a missed case is a silent
startup, never a false alarm.
## Still in the program

27
fastmcp_slim/fastmcp/server/low_level.py
View file

@ -181,6 +181,7 @@ class FastMCPServerMiddleware:
self, ctx: ServerRequestContext, call_next: CallNext
) -> HandlerResult:
from fastmcp.server.dependencies import bind_request_context
from fastmcp.server.protocol_versions import protocol_version_error
fastmcp = self._ref()
with (
@ -192,6 +193,18 @@ class FastMCPServerMiddleware:
return await call_next(ctx)
if ctx.method == "initialize" and ctx.request_id is not None:
return await self._run_initialize_mw(fastmcp, ctx, call_next)
# Refuse a request whose protocol version the server does not serve.
# The handshake is vetoed at `initialize` above; a modern connection
# has no handshake to veto (a client pinned to a modern version can
# skip `server/discover` entirely), so the request itself is the
# enforcement point. Routing the refusal through the outer pass lets
# `on_message`/`on_request` observe it like any other early failure.
if ctx.request_id is not None:
version_error = protocol_version_error(fastmcp, ctx.protocol_version)
if version_error is not None:
return await self._run_outer_mw(
fastmcp, ctx, call_next, _raise=version_error
)
if ctx.request_id is not None and ctx.method in _INTERIOR_METHODS:
return await self._dispatch_component(fastmcp, ctx, call_next)
return await self._run_outer_mw(fastmcp, ctx, call_next, _raise=None)
@ -326,7 +339,9 @@ class FastMCPServerMiddleware:
) -> HandlerResult:
from fastmcp.server.context import Context
from fastmcp.server.middleware.middleware import MiddlewareContext
from fastmcp.server.protocol_floor import enforce_handshake_floor
from fastmcp.server.protocol_versions import (
enforce_handshake_protocol_version,
)
# Reconstruct the InitializeRequest from the raw params so FastMCP
# middleware `on_initialize` hooks that inspect the message still work.
@ -350,11 +365,11 @@ class FastMCPServerMiddleware:
_mw_ctx: MiddlewareContext,
) -> mcp_types.InitializeResult | None:
# Refuse the handshake before it commits when the negotiated version
# is below the server's declared protocol floor. Raising MCPError
# here (before call_next) vetoes initialize on the framework-owned
# path, so the client sees a clear connect-time refusal instead of a
# runtime era error mid tool-call.
enforce_handshake_floor(fastmcp, init_message)
# is not one the server serves. Raising MCPError here (before
# call_next) vetoes initialize on the framework-owned path, so the
# client sees a clear connect-time refusal instead of a runtime era
# error mid tool-call.
enforce_handshake_protocol_version(fastmcp, init_message)
# call_next(ctx) runs the rest of the SDK chain, which for
# initialize returns the serialized InitializeResult dict. FastMCP
# middleware `on_initialize` hooks expect a typed InitializeResult,

11
fastmcp_slim/fastmcp/server/mixins/lifespan.py
View file

@ -246,11 +246,12 @@ class LifespanMixin:
for provider in self.providers:
await stack.enter_async_context(provider.lifespan())
# Warn (never raise) when the declared protocol floor and the
# registered features are incoherent — e.g. modern-only guard tools
# with no modern floor. Runs once per fresh lifespan entry, after all
# providers are mounted so their components are visible.
from fastmcp.server.protocol_floor import check_protocol_coherence
# Warn (never raise) when a declared `protocol_versions` set cannot
# carry a capability the server actually uses — e.g. modern-only
# guard tools under a handshake-only declaration. Silent when the
# server declared nothing. Runs once per fresh lifespan entry, after
# all providers are mounted so their components are visible.
from fastmcp.server.protocol_versions import check_protocol_coherence
await check_protocol_coherence(self)

210
fastmcp_slim/fastmcp/server/protocol_floor.py
View file

@ -1,210 +0,0 @@
"""Server protocol-version floor: declaration, negotiation-time enforcement, and
startup incoherence detection.
A FastMCP server may depend on features that only exist on a particular MCP
protocol era. The clearest example is the modern (``2026-07-28``) multi-round
"guard" pattern (SEP-2322): a tool that returns an ``InputRequiredResult`` works
only on a modern connection. A legacy client that reaches such a tool over the
initialize handshake gets a confusing era error *mid tool-call* instead of a
clear refusal at connect time.
This module lets a server declare a minimum protocol version (a "floor") and
enforces it at the two connection-negotiation points FastMCP owns:
* the initialize handshake (legacy era), refused before the handshake commits;
* ``server/discover`` (modern era), which always satisfies any currently
declarable floor because the modern era is the newest era.
It also runs a conservative, warning-only coherence check at server startup: it
infers the modern requirement from registered guard tools and flags declared
floors that contradict a configured back-channel handler.
.. note::
The public spelling (``FastMCP(min_protocol_version=...)``) is **provisional**
and expected to change. The negotiation hook and the inference rules in this
module are valid under any eventual spelling; only the constructor keyword is
a placeholder.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from mcp.shared.exceptions import MCPError
from mcp_types import INVALID_PARAMS
from mcp_types.version import (
HANDSHAKE_PROTOCOL_VERSIONS,
KNOWN_PROTOCOL_VERSIONS,
LATEST_HANDSHAKE_VERSION,
MODERN_PROTOCOL_VERSIONS,
is_version_at_least,
)
from fastmcp.tools.function_parsing import _contains_input_required
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
import mcp_types
from fastmcp.server.server import FastMCP
from fastmcp.tools.base import Tool
logger = get_logger(__name__)
# The oldest modern (per-request-envelope) protocol version. A floor at or above
# this value means "modern connections only" — no handshake-era client can
# satisfy it, since the handshake era tops out at LATEST_HANDSHAKE_VERSION.
_MODERN_FLOOR = MODERN_PROTOCOL_VERSIONS[0]
def validate_protocol_floor(value: str | None) -> str | None:
"""Validate a declared protocol-version floor at construction time.
Returns the value unchanged when it is ``None`` (no floor) or a known
protocol revision. Raises ``ValueError`` for any unrecognized string — a
floor the SDK could never negotiate is a programming error, not a runtime
condition to warn about.
"""
if value is None:
return None
if value not in KNOWN_PROTOCOL_VERSIONS:
raise ValueError(
f"min_protocol_version={value!r} is not a known MCP protocol version. "
f"Known versions: {', '.join(KNOWN_PROTOCOL_VERSIONS)}."
)
return value
def handshake_negotiated_version(requested: str | None) -> str:
"""The version an initialize handshake would settle on for ``requested``.
Mirrors the SDK's ``ServerRunner._negotiate_initialize``: a client's
requested handshake revision is honored; anything else (an unknown string,
or a modern-era version the handshake cannot serve) counters with the newest
handshake revision. The connection operates at the returned version, so it is
what the floor check must compare against.
"""
if requested is not None and requested in HANDSHAKE_PROTOCOL_VERSIONS:
return requested
return LATEST_HANDSHAKE_VERSION
def enforce_handshake_floor(
fastmcp: FastMCP,
init_message: mcp_types.InitializeRequest | None,
) -> None:
"""Refuse an initialize handshake that cannot meet the server's floor.
Called from the framework-owned initialize path (the SDK's middleware layer)
before the handshake commits. When the connection's negotiated handshake
version is below the floor, raises ``MCPError`` so the client sees a clear
connect-time refusal naming the required version instead of a runtime era
error later. A ``None`` floor (the default) never refuses.
"""
floor = fastmcp.min_protocol_version
if floor is None or init_message is None:
return
requested = init_message.params.protocol_version
negotiated = handshake_negotiated_version(requested)
if is_version_at_least(negotiated, floor):
return
detail = (
"Connect using the modern protocol (server/discover) instead."
if is_version_at_least(floor, _MODERN_FLOOR)
else "Upgrade the client or connect with a newer protocol version."
)
raise MCPError(
code=INVALID_PARAMS,
message=(
f"Server {fastmcp.name!r} requires MCP protocol version {floor} or "
f"newer; the initialize handshake offered {requested!r} "
f"(negotiates to {negotiated}). {detail}"
),
data={"requiredProtocolVersion": floor, "offeredProtocolVersion": requested},
)
def tool_requires_modern(tool: Tool) -> bool:
"""True when a tool's return annotation makes it a modern-only guard tool.
A guard tool (SEP-2322) returns an ``InputRequiredResult`` to ask the client
for input across rounds; that pattern exists only on the modern era. Detection
reuses ``_contains_input_required`` over the tool's captured return annotation,
so every union/alias/``Annotated`` shape the parser recognizes is covered. Only
``FunctionTool`` carries a return annotation; other tool kinds return ``False``
(a conservative miss, not a false positive).
"""
if not isinstance(tool, FunctionTool):
return False
return _contains_input_required(tool.return_type)
async def check_protocol_coherence(fastmcp: FastMCP) -> None:
"""Warn at startup when the declared floor and registered features conflict.
Conservative by design: emits actionable warnings, never raises. Two rules:
1. **Modern-only guard tools under a non-modern floor.** If any registered
tool is a guard tool (returns ``InputRequiredResult``) but the floor does
not guarantee a modern connection, handshake-era clients that reach those
tools fail mid-call. Recommends declaring a modern floor.
2. **Modern floor with a back-channel sampling fallback.** A modern floor
forbids the server-initiated back-channel, so a ``sampling_handler`` set to
``"fallback"`` (prefer the client's model, fall back to the local handler)
can never actually reach the client — the local handler always runs. Flags
the dead preference.
Runtime-only back-channel usage (``ctx.elicit`` / ``ctx.sample`` /
``ctx.list_roots``) has no reliable static signal, so it is deliberately not
inferred here.
"""
floor = fastmcp.min_protocol_version
floor_is_modern = floor is not None and is_version_at_least(floor, _MODERN_FLOOR)
# Inspect only this server's directly-registered tools. Aggregating mounted
# children would route through their middleware chains (a startup side
# effect); mounted or transformed guard tools are a conservative miss, not a
# false positive. `LocalProvider.list_tools` is side-effect-free.
try:
tools = list(await fastmcp._local_provider.list_tools())
except Exception as exc:
logger.debug("Protocol coherence check could not list tools: %s", exc)
tools = []
if not floor_is_modern:
guard_tools = sorted(t.name for t in tools if tool_requires_modern(t))
if guard_tools:
floor_desc = (
"no minimum protocol version is declared"
if floor is None
else f"the declared floor is {floor}"
)
logger.warning(
"Server %r registers guard tool(s) %s that return "
"InputRequiredResult and require the modern MCP protocol "
"(%s), but %s. Handshake-era clients calling these tools will "
"fail mid-call. Declare min_protocol_version=%r to refuse such "
"clients at connect time.",
fastmcp.name,
", ".join(guard_tools),
_MODERN_FLOOR,
floor_desc,
_MODERN_FLOOR,
)
if (
floor_is_modern
and fastmcp.sampling_handler is not None
and fastmcp.sampling_handler_behavior == "fallback"
):
logger.warning(
"Server %r declares a modern protocol floor (%s) but configures a "
"sampling_handler with behavior 'fallback'. The modern protocol "
"forbids the server-initiated back-channel, so the fallback never "
"reaches the client and the local handler always runs. Use "
"sampling_handler_behavior='always' if that is intended, or lower "
"the floor to allow the client back-channel.",
fastmcp.name,
floor,
)

320
fastmcp_slim/fastmcp/server/protocol_versions.py Normal file
View file

@ -0,0 +1,320 @@
"""Server protocol-version restriction: declaration, enforcement, and startup
coherence checks.
A FastMCP server may depend on features that exist on only one MCP protocol era.
The modern (``2026-07-28``) era added the multi-round-trip "guard" pattern
(SEP-2322) and *removed* the server-initiated back-channel (``ctx.elicit``,
``ctx.sample``, ``ctx.list_roots``) that the handshake eras provide. Neither era
is a superset of the other, so a server can legitimately require either one.
Protocol versions are therefore modeled the way the SDK models them — as *an
enumerated set, not an ordered scalar* (see ``mcp_types.version``). A server
declares the set of protocol versions it is willing to serve, and enforcement is
set membership:
from mcp_types.version import MODERN_PROTOCOL_VERSIONS
FastMCP("guarded", protocol_versions=MODERN_PROTOCOL_VERSIONS)
The SDK's era tuples are the durable way to name an era: they grow on their own
when the SDK adds a revision to an era, so a server pinned to
``MODERN_PROTOCOL_VERSIONS`` keeps working across SDK upgrades without FastMCP
inventing alias vocabulary of its own.
Enforcement happens at both connection paths FastMCP owns:
* the initialize handshake, refused before the handshake commits;
* every modern per-request envelope, refused on the request itself. A client
pinned to a modern version never probes ``server/discover``, so refusing
discovery alone would not cover it — the request *is* the connection.
Refusals use the spec-standard ``-32022`` unsupported-protocol-version error
carrying the server's supported list, so a negotiating (``mode="auto"``) client
reads it as guidance rather than as a dead end: refused at ``server/discover``
by a handshake-only server, it sees handshake versions in ``supported`` and
falls back to the initialize handshake on its own.
Declaring nothing (the default) serves every era the SDK supports.
"""
from __future__ import annotations
from collections.abc import Callable, Iterable, Sequence
from dataclasses import dataclass
from typing import TYPE_CHECKING
from mcp.shared.exceptions import MCPError
from mcp_types import (
UNSUPPORTED_PROTOCOL_VERSION,
UnsupportedProtocolVersionErrorData,
)
from mcp_types.version import (
HANDSHAKE_PROTOCOL_VERSIONS,
KNOWN_PROTOCOL_VERSIONS,
LATEST_HANDSHAKE_VERSION,
MODERN_PROTOCOL_VERSIONS,
)
from fastmcp.tools.function_parsing import _contains_input_required
from fastmcp.tools.function_tool import FunctionTool
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
import mcp_types
from fastmcp.server.server import FastMCP
from fastmcp.tools.base import Tool
logger = get_logger(__name__)
def validate_protocol_versions(
value: Iterable[str] | None,
) -> tuple[str, ...] | None:
"""Validate and normalize a declared set of protocol versions.
Returns ``None`` unchanged (no restriction), or the declared versions as a
deduplicated tuple in SDK order. Raises ``ValueError`` for an empty set (a
server that serves no protocol version can never be reached) or for any
string the SDK does not recognize — a version the SDK could never negotiate
is a programming error, not a runtime condition to warn about.
"""
if value is None:
return None
declared = list(value)
unknown = [v for v in declared if v not in KNOWN_PROTOCOL_VERSIONS]
if unknown:
raise ValueError(
f"protocol_versions contains unknown MCP protocol version(s): "
f"{', '.join(repr(v) for v in unknown)}. "
f"Known versions: {', '.join(KNOWN_PROTOCOL_VERSIONS)}. "
f"Prefer the SDK's era tuples "
f"(mcp_types.version.MODERN_PROTOCOL_VERSIONS / "
f"HANDSHAKE_PROTOCOL_VERSIONS) over literal strings."
)
normalized = tuple(v for v in KNOWN_PROTOCOL_VERSIONS if v in declared)
if not normalized:
raise ValueError(
"protocol_versions must name at least one protocol version. "
"Pass None (the default) to serve every protocol version."
)
return normalized
def describe_protocol_versions(versions: Sequence[str]) -> str:
"""Human-readable description of a declared version set for error messages."""
if tuple(versions) == MODERN_PROTOCOL_VERSIONS:
return (
f"the modern protocol ({', '.join(versions)}), reached via server/discover"
)
if tuple(versions) == HANDSHAKE_PROTOCOL_VERSIONS:
return f"the handshake protocol ({', '.join(versions)}), reached via initialize"
return ", ".join(versions)
def _remedy_for(versions: Sequence[str]) -> str:
"""Actionable next step for a client that failed the version check."""
wants_modern = any(v in MODERN_PROTOCOL_VERSIONS for v in versions)
wants_handshake = any(v in HANDSHAKE_PROTOCOL_VERSIONS for v in versions)
if wants_modern and not wants_handshake:
return "Connect using the modern protocol (server/discover) instead."
if wants_handshake and not wants_modern:
return "Connect using the initialize handshake instead."
return "Use one of the protocol versions this server serves."
def protocol_version_error(fastmcp: FastMCP, version: str) -> MCPError | None:
"""The refusal for ``version``, or ``None`` when the server serves it.
A server that declared nothing (the default) serves every version and never
refuses. Otherwise this is plain set membership: the declared versions are a
set, not a bound, so a handshake-only server refuses modern connections just
as a modern-only server refuses handshake connections.
The refusal is the spec-standard ``-32022`` unsupported-protocol-version
error carrying the server's supported list, which is what a negotiating
client already knows how to read: an ``auto`` client refused at
``server/discover`` sees handshake versions in ``supported`` and falls back
to the initialize handshake instead of failing the connect.
"""
allowed = fastmcp.protocol_versions
if allowed is None or version in allowed:
return None
return MCPError(
code=UNSUPPORTED_PROTOCOL_VERSION,
message=(
f"Server {fastmcp.name!r} serves {describe_protocol_versions(allowed)}; "
f"this connection uses MCP protocol version {version}. "
f"{_remedy_for(allowed)}"
),
data=UnsupportedProtocolVersionErrorData(
supported=list(allowed), requested=version
).model_dump(by_alias=True, mode="json"),
)
def handshake_negotiated_version(requested: str | None) -> str:
"""The version an initialize handshake would settle on for ``requested``.
Mirrors the SDK's ``ServerRunner._negotiate_initialize``: a client's
requested handshake revision is honored; anything else (an unknown string,
or a modern-era version the handshake cannot serve) counters with the newest
handshake revision. The connection operates at the returned version, so it is
what the membership check must compare against.
"""
if requested is not None and requested in HANDSHAKE_PROTOCOL_VERSIONS:
return requested
return LATEST_HANDSHAKE_VERSION
def enforce_handshake_protocol_version(
fastmcp: FastMCP,
init_message: mcp_types.InitializeRequest | None,
) -> None:
"""Refuse an initialize handshake the server does not serve.
Called from the framework-owned initialize path before the handshake
commits, so the client sees a clear connect-time refusal naming what the
server serves instead of a confusing era error mid tool-call.
"""
if fastmcp.protocol_versions is None or init_message is None:
return
requested = init_message.params.protocol_version
error = protocol_version_error(fastmcp, handshake_negotiated_version(requested))
if error is not None:
raise error
# ---------------------------------------------------------------------------
# Capability map
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class ProtocolCapability:
"""A server-side capability that only some protocol versions can carry.
``detect`` returns evidence (tool names, a configuration description) that
the server actually uses the capability, and is consulted only when the
declared version set cannot carry it — so detection never costs anything on
a server that declared nothing.
"""
key: str
label: str
versions: tuple[str, ...]
remedy: str
detect: Callable[[FastMCP, Sequence[Tool]], list[str]]
def tool_uses_multi_round_trip(tool: Tool) -> bool:
"""True when a tool's return annotation makes it a modern-only guard tool.
A guard tool (SEP-2322) returns an ``InputRequiredResult`` to ask the client
for input across rounds; that pattern exists only on the modern era.
Detection reuses ``_contains_input_required`` over the tool's captured return
annotation, so every union/alias/``Annotated`` shape the parser recognizes is
covered. Only ``FunctionTool`` carries a return annotation; other tool kinds
return ``False`` (a conservative miss, not a false positive).
"""
if not isinstance(tool, FunctionTool):
return False
return _contains_input_required(tool.return_type)
def _detect_multi_round_trip(fastmcp: FastMCP, tools: Sequence[Tool]) -> list[str]:
return sorted(t.name for t in tools if tool_uses_multi_round_trip(t))
def _detect_client_back_channel(fastmcp: FastMCP, tools: Sequence[Tool]) -> list[str]:
"""Configuration-level evidence that the server wants the client back-channel.
``ctx.elicit`` / ``ctx.sample`` / ``ctx.list_roots`` are *runtime* calls with
no reliable static signal, so only configuration contradictions are
detectable. A ``sampling_handler`` set to ``"fallback"`` says "prefer the
client's model, fall back to mine" — a preference that cannot be honored on a
protocol version without the back-channel.
"""
if (
fastmcp.sampling_handler is not None
and fastmcp.sampling_handler_behavior == "fallback"
):
return ["sampling_handler with behavior='fallback'"]
return []
PROTOCOL_CAPABILITIES: tuple[ProtocolCapability, ...] = (
ProtocolCapability(
key="multi_round_trip",
label="multi-round-trip guard tools (they return InputRequiredResult)",
versions=MODERN_PROTOCOL_VERSIONS,
remedy=(
"Include the modern versions in protocol_versions "
"(mcp_types.version.MODERN_PROTOCOL_VERSIONS), or stop returning "
"InputRequiredResult from these tools."
),
detect=_detect_multi_round_trip,
),
ProtocolCapability(
key="client_back_channel",
label=(
"the server-initiated client back-channel "
"(ctx.elicit / ctx.sample / ctx.list_roots)"
),
versions=HANDSHAKE_PROTOCOL_VERSIONS,
remedy=(
"Include the handshake versions in protocol_versions "
"(mcp_types.version.HANDSHAKE_PROTOCOL_VERSIONS), or use "
"sampling_handler_behavior='always' so the local handler is the "
"intended path."
),
detect=_detect_client_back_channel,
),
# Extension point: a capability that only some protocol versions can carry
# is one entry here, not a new special case in the check below. The 2026
# tasks extension is the next expected entry — when FastMCP implements it,
# add a ProtocolCapability naming the versions that carry it and a `detect`
# that reports the registered task-enabled components.
)
async def check_protocol_coherence(fastmcp: FastMCP) -> None:
"""Warn at startup when a declared version set contradicts what is registered.
Warning-only by design, and **silent unless the server declared a version
set**. A server with an ordinary mix of tools is fine: a guard tool raises a
clear era error if an old client reaches for it, and warning about that at
startup would train people to ignore warnings. The check fires only when the
author asserted something and the registration contradicts the assertion.
"""
allowed = fastmcp.protocol_versions
if allowed is None:
return
unmet = [c for c in PROTOCOL_CAPABILITIES if set(allowed).isdisjoint(c.versions)]
if not unmet:
return
# Inspect only this server's directly-registered tools. Aggregating mounted
# children would route through their middleware chains (a startup side
# effect); mounted or transformed components are a conservative miss, not a
# false positive. `LocalProvider.list_tools` is side-effect-free.
try:
tools: list[Tool] = list(await fastmcp._local_provider.list_tools())
except (LookupError, RuntimeError, ValueError) as exc:
logger.debug("Protocol coherence check could not list tools: %s", exc)
tools = []
for capability in unmet:
evidence = capability.detect(fastmcp, tools)
if not evidence:
continue
logger.warning(
"Server %r declares protocol_versions=%s, which cannot carry %s, "
"but the server uses it: %s. %s",
fastmcp.name,
list(allowed),
capability.label,
", ".join(evidence),
capability.remedy,
)

33
fastmcp_slim/fastmcp/server/server.py
View file

@ -8,6 +8,7 @@ import secrets
from collections.abc import (
AsyncIterator,
Callable,
Iterable,
Sequence,
)
from contextlib import (
@ -77,7 +78,7 @@ from fastmcp.server.middleware.middleware import (
mark_interior_dispatched,
)
from fastmcp.server.mixins import LifespanMixin, MCPOperationsMixin, TransportMixin
from fastmcp.server.protocol_floor import validate_protocol_floor
from fastmcp.server.protocol_versions import validate_protocol_versions
from fastmcp.server.providers import LocalProvider, Provider
from fastmcp.server.providers.aggregate import AggregateProvider
from fastmcp.server.tasks.config import TaskConfig, TaskMeta
@ -357,7 +358,7 @@ class FastMCP(
session_state_store: AsyncKeyValue | None = None,
sampling_handler: SamplingHandler | None = None,
sampling_handler_behavior: Literal["always", "fallback"] | None = None,
min_protocol_version: str | None = None,
protocol_versions: Iterable[str] | None = None,
client_log_level: mcp_types.LoggingLevel | None = None,
experimental_capabilities: dict[str, dict[str, Any]] | None = None,
**kwargs: Any,
@ -532,16 +533,14 @@ class FastMCP(
sampling_handler_behavior or "fallback"
)
# Minimum MCP protocol version this server requires. Enforced at
# connection negotiation (the initialize handshake refuses clients below
# the floor) and checked for coherence against registered features at
# startup. `None` (the default) declares no floor: the server serves
# every protocol era, fully backward compatible.
#
# NOTE: the `min_protocol_version` spelling is provisional; see
# `fastmcp.server.protocol_floor`.
self._min_protocol_version: str | None = validate_protocol_floor(
min_protocol_version
# The set of MCP protocol versions this server is willing to serve.
# Enforced by set membership at both connection paths (the initialize
# handshake and the modern per-request envelope) and checked for
# coherence against registered features at startup. `None` (the default)
# declares no restriction: the server serves every protocol version the
# SDK supports, fully backward compatible.
self._protocol_versions: tuple[str, ...] | None = validate_protocol_versions(
protocol_versions
)
def __repr__(self) -> str:
@ -564,13 +563,13 @@ class FastMCP(
return self._mcp_server.version
@property
def min_protocol_version(self) -> str | None:
"""The minimum MCP protocol version this server requires, if declared.
def protocol_versions(self) -> tuple[str, ...] | None:
"""The MCP protocol versions this server serves, if restricted.
`None` means no floor (serve every protocol era). The spelling is
provisional; see `fastmcp.server.protocol_floor`.
`None` (the default) means no restriction: every protocol version the
SDK supports is served. Otherwise the declared versions, in SDK order.
"""
return self._min_protocol_version
return self._protocol_versions
@property
def website_url(self) -> str | None:

289
tests/server/test_protocol_floor.py
View file

@ -1,289 +0,0 @@
"""Server protocol-version floor: declaration, negotiation-time enforcement, and
startup incoherence detection.
A server may declare a minimum protocol version. The initialize handshake refuses
clients below the floor before the handshake commits; the modern
(``server/discover``) era always satisfies any currently declarable floor. A
conservative startup check warns when the declared floor and the registered
features are incoherent.
The ``min_protocol_version`` spelling is provisional; these tests exercise the
mechanics, which hold under any eventual spelling.
"""
from __future__ import annotations
import logging
import mcp_types
import pytest
from exceptiongroup import BaseExceptionGroup
from mcp.client import Client as SDKClient
from mcp.server import Server as LowLevelServer
from mcp.shared.exceptions import MCPError
from fastmcp import Context, FastMCP
from fastmcp.server.protocol_floor import (
handshake_negotiated_version,
tool_requires_modern,
validate_protocol_floor,
)
from fastmcp.tools.base import Tool
_COHERENCE_LOGGER = "fastmcp.server.protocol_floor"
def _server(mcp: FastMCP) -> LowLevelServer:
"""The lowlevel Server the SDK client connects to in-process."""
return mcp._mcp_server
def _find_mcp_error(exc: BaseException) -> MCPError | None:
"""Unwrap the MCPError a refused in-memory handshake surfaces.
The legacy in-memory transport runs ``initialize`` inside a task group, so a
connect-time refusal propagates as an ``ExceptionGroup`` wrapping the
``MCPError`` rather than the bare error.
"""
if isinstance(exc, MCPError):
return exc
if isinstance(exc, BaseExceptionGroup):
for inner in exc.exceptions:
found = _find_mcp_error(inner)
if found is not None:
return found
if exc.__cause__ is not None:
return _find_mcp_error(exc.__cause__)
return None
# ---------------------------------------------------------------------------
# Construction-time validation
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"version",
["2024-11-05", "2025-06-18", "2025-11-25", "2026-07-28", None],
)
def test_valid_floor_accepted(version):
assert validate_protocol_floor(version) == version
assert FastMCP("s", min_protocol_version=version).min_protocol_version == version
@pytest.mark.parametrize("version", ["9999-01-01", "latest", "2026", ""])
def test_unknown_floor_rejected(version):
with pytest.raises(ValueError, match="not a known MCP protocol version"):
validate_protocol_floor(version)
with pytest.raises(ValueError, match="not a known MCP protocol version"):
FastMCP("s", min_protocol_version=version)
def test_default_is_no_floor():
assert FastMCP("s").min_protocol_version is None
# ---------------------------------------------------------------------------
# Handshake negotiation mirror
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"requested, expected",
[
("2025-11-25", "2025-11-25"),
("2025-06-18", "2025-06-18"),
("2024-11-05", "2024-11-05"),
# A modern-era or unknown request counters with the newest handshake.
("2026-07-28", "2025-11-25"),
("garbage", "2025-11-25"),
(None, "2025-11-25"),
],
)
def test_handshake_negotiated_version(requested, expected):
assert handshake_negotiated_version(requested) == expected
# ---------------------------------------------------------------------------
# Guard-tool detection
# ---------------------------------------------------------------------------
def test_guard_tool_detected():
def guard(x: int) -> str | mcp_types.InputRequiredResult:
return "ok"
assert tool_requires_modern(Tool.from_function(guard)) is True
def test_plain_tool_not_flagged():
def plain(x: int) -> int:
return x
assert tool_requires_modern(Tool.from_function(plain)) is False
# ---------------------------------------------------------------------------
# Negotiation-time enforcement (handshake path)
# ---------------------------------------------------------------------------
@pytest.fixture
def floored_server() -> FastMCP:
mcp = FastMCP("floored", min_protocol_version="2026-07-28")
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
return mcp
async def test_modern_floor_refuses_legacy_handshake(floored_server):
with pytest.raises(BaseException) as excinfo:
async with SDKClient(_server(floored_server), mode="legacy") as client:
await client.list_tools()
err = _find_mcp_error(excinfo.value)
assert err is not None
assert err.code == mcp_types.INVALID_PARAMS
assert "2026-07-28" in err.message
assert "server/discover" in err.message
@pytest.mark.parametrize("mode", ["auto", "2026-07-28"])
async def test_modern_floor_allows_modern(floored_server, mode):
async with SDKClient(_server(floored_server), mode=mode) as client:
result = await client.list_tools()
assert [t.name for t in result.tools] == ["add"]
async def test_no_floor_allows_legacy_handshake():
mcp = FastMCP("open")
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
async with SDKClient(_server(mcp), mode="legacy") as client:
assert client.protocol_version == "2025-11-25"
result = await client.list_tools()
assert [t.name for t in result.tools] == ["add"]
async def test_handshake_floor_allows_equal_version_client():
mcp = FastMCP("hs", min_protocol_version="2025-11-25")
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
async with SDKClient(_server(mcp), mode="legacy") as client:
assert client.protocol_version == "2025-11-25"
result = await client.list_tools()
assert [t.name for t in result.tools] == ["add"]
# ---------------------------------------------------------------------------
# Startup incoherence detection (warnings only)
# ---------------------------------------------------------------------------
def _coherence_warnings(caplog) -> list[str]:
return [
r.getMessage()
for r in caplog.records
if r.name == _COHERENCE_LOGGER and r.levelno == logging.WARNING
]
async def test_guard_tool_without_floor_warns(caplog):
mcp = FastMCP("guardy")
@mcp.tool
def ask(x: int) -> str | mcp_types.InputRequiredResult:
return "ok"
with caplog.at_level(logging.WARNING, logger=_COHERENCE_LOGGER):
async with mcp._lifespan_manager():
pass
warnings = _coherence_warnings(caplog)
assert any("ask" in w and "modern" in w.lower() for w in warnings)
async def test_guard_tool_with_modern_floor_silent(caplog):
mcp = FastMCP("guardy", min_protocol_version="2026-07-28")
@mcp.tool
def ask(x: int) -> str | mcp_types.InputRequiredResult:
return "ok"
with caplog.at_level(logging.WARNING, logger=_COHERENCE_LOGGER):
async with mcp._lifespan_manager():
pass
assert _coherence_warnings(caplog) == []
async def test_modern_floor_with_fallback_sampling_warns(caplog):
async def handler(messages, params, context):
return "x"
mcp = FastMCP(
"samp",
min_protocol_version="2026-07-28",
sampling_handler=handler,
sampling_handler_behavior="fallback",
)
with caplog.at_level(logging.WARNING, logger=_COHERENCE_LOGGER):
async with mcp._lifespan_manager():
pass
warnings = _coherence_warnings(caplog)
assert any("fallback" in w and "back-channel" in w for w in warnings)
async def test_modern_floor_with_always_sampling_silent(caplog):
async def handler(messages, params, context):
return "x"
mcp = FastMCP(
"samp",
min_protocol_version="2026-07-28",
sampling_handler=handler,
sampling_handler_behavior="always",
)
with caplog.at_level(logging.WARNING, logger=_COHERENCE_LOGGER):
async with mcp._lifespan_manager():
pass
assert _coherence_warnings(caplog) == []
async def test_plain_server_is_coherent(caplog):
mcp = FastMCP("clean")
@mcp.tool
def plain(a: int) -> int:
return a
with caplog.at_level(logging.WARNING, logger=_COHERENCE_LOGGER):
async with mcp._lifespan_manager():
pass
assert _coherence_warnings(caplog) == []
async def test_guard_tool_reaches_modern_client(floored_server):
"""A guard tool served under a modern floor works end-to-end on modern."""
mcp = FastMCP("guarded", min_protocol_version="2026-07-28")
@mcp.tool
async def confirm(ctx: Context) -> str | mcp_types.InputRequiredResult:
return "confirmed"
async with SDKClient(_server(mcp), mode="auto") as client:
result = await client.call_tool("confirm", {})
assert result.is_error is False

451
tests/server/test_protocol_versions.py Normal file
View file

@ -0,0 +1,451 @@
"""Server protocol-version restriction: declaration, enforcement, and startup
coherence checks.
A server declares the *set* of protocol versions it serves. Membership — not
ordering — decides whether a connection is accepted, so a handshake-only server
refuses modern connections just as a modern-only server refuses handshake
connections. A startup check warns (never raises) when a declared set cannot
carry a capability the server actually uses, and stays silent when nothing was
declared.
"""
from __future__ import annotations
import logging
import mcp_types
import pytest
from exceptiongroup import BaseExceptionGroup
from mcp.client import Client as SDKClient
from mcp.server import Server as LowLevelServer
from mcp.shared.exceptions import MCPError
from mcp_types.version import (
HANDSHAKE_PROTOCOL_VERSIONS,
MODERN_PROTOCOL_VERSIONS,
)
from fastmcp import Client, Context, FastMCP
from fastmcp.server.protocol_versions import (
enforce_handshake_protocol_version,
handshake_negotiated_version,
tool_uses_multi_round_trip,
validate_protocol_versions,
)
from fastmcp.tools.base import Tool
_COHERENCE_LOGGER = "fastmcp.server.protocol_versions"
def _server(mcp: FastMCP) -> LowLevelServer:
"""The lowlevel Server the SDK client connects to in-process."""
return mcp._mcp_server
def _find_mcp_error(exc: BaseException) -> MCPError | None:
"""Unwrap the MCPError a refused in-memory connection surfaces.
The legacy in-memory transport runs ``initialize`` inside a task group, so a
connect-time refusal propagates as an ``ExceptionGroup`` wrapping the
``MCPError`` rather than the bare error.
"""
if isinstance(exc, MCPError):
return exc
if isinstance(exc, BaseExceptionGroup):
for inner in exc.exceptions:
found = _find_mcp_error(inner)
if found is not None:
return found
if exc.__cause__ is not None:
return _find_mcp_error(exc.__cause__)
return None
# ---------------------------------------------------------------------------
# Construction-time validation
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"declared, expected",
[
(None, None),
(MODERN_PROTOCOL_VERSIONS, ("2026-07-28",)),
(HANDSHAKE_PROTOCOL_VERSIONS, HANDSHAKE_PROTOCOL_VERSIONS),
(["2026-07-28"], ("2026-07-28",)),
(["2025-06-18", "2026-07-28"], ("2025-06-18", "2026-07-28")),
# Normalized to SDK order, deduplicated.
(["2026-07-28", "2024-11-05", "2026-07-28"], ("2024-11-05", "2026-07-28")),
# Any iterable, not just a sequence.
({"2025-11-25"}, ("2025-11-25",)),
],
)
def test_valid_protocol_versions_normalized(declared, expected):
assert validate_protocol_versions(declared) == expected
assert FastMCP("s", protocol_versions=declared).protocol_versions == expected
@pytest.mark.parametrize(
"declared",
[["9999-01-01"], ["modern"], ["2026"], [""], ["2026-07-28", "handshake"]],
)
def test_unknown_protocol_version_rejected(declared):
with pytest.raises(ValueError, match="unknown MCP protocol version"):
validate_protocol_versions(declared)
with pytest.raises(ValueError, match="unknown MCP protocol version"):
FastMCP("s", protocol_versions=declared)
def test_empty_protocol_versions_rejected():
with pytest.raises(ValueError, match="at least one protocol version"):
validate_protocol_versions([])
with pytest.raises(ValueError, match="at least one protocol version"):
FastMCP("s", protocol_versions=[])
def test_default_serves_every_version():
assert FastMCP("s").protocol_versions is None
# ---------------------------------------------------------------------------
# Handshake negotiation mirror
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"requested, expected",
[
("2025-11-25", "2025-11-25"),
("2025-06-18", "2025-06-18"),
("2024-11-05", "2024-11-05"),
# A modern-era or unknown request counters with the newest handshake.
("2026-07-28", "2025-11-25"),
("garbage", "2025-11-25"),
(None, "2025-11-25"),
],
)
def test_handshake_negotiated_version(requested, expected):
assert handshake_negotiated_version(requested) == expected
# ---------------------------------------------------------------------------
# Guard-tool detection
# ---------------------------------------------------------------------------
def test_guard_tool_detected():
def guard(x: int) -> str | mcp_types.InputRequiredResult:
return "ok"
assert tool_uses_multi_round_trip(Tool.from_function(guard)) is True
def test_plain_tool_not_flagged():
def plain(x: int) -> int:
return x
assert tool_uses_multi_round_trip(Tool.from_function(plain)) is False
# ---------------------------------------------------------------------------
# Enforcement: modern-only server refuses handshake clients
# ---------------------------------------------------------------------------
@pytest.fixture
def modern_only_server() -> FastMCP:
mcp = FastMCP("modern-only", protocol_versions=MODERN_PROTOCOL_VERSIONS)
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
return mcp
async def test_modern_only_refuses_handshake(modern_only_server):
with pytest.raises(BaseException) as excinfo:
async with SDKClient(_server(modern_only_server), mode="legacy") as client:
await client.list_tools()
err = _find_mcp_error(excinfo.value)
assert err is not None
assert err.code == mcp_types.UNSUPPORTED_PROTOCOL_VERSION
assert "2026-07-28" in err.message
assert "server/discover" in err.message
@pytest.mark.parametrize("mode", ["auto", "2026-07-28"])
async def test_modern_only_allows_modern(modern_only_server, mode):
async with SDKClient(_server(modern_only_server), mode=mode) as client:
result = await client.list_tools()
assert [t.name for t in result.tools] == ["add"]
# ---------------------------------------------------------------------------
# Enforcement: handshake-only server refuses modern clients
#
# This case is only expressible because the declaration is a set, not a bound:
# under a minimum-version model there was no way to say "the session era".
# ---------------------------------------------------------------------------
@pytest.fixture
def handshake_only_server() -> FastMCP:
mcp = FastMCP("handshake-only", protocol_versions=HANDSHAKE_PROTOCOL_VERSIONS)
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
return mcp
async def test_handshake_only_allows_handshake(handshake_only_server):
async with SDKClient(_server(handshake_only_server), mode="legacy") as client:
assert client.protocol_version == "2025-11-25"
result = await client.list_tools()
assert [t.name for t in result.tools] == ["add"]
async def test_handshake_only_refuses_pinned_modern_client(handshake_only_server):
"""A client pinned to a modern version never probes discover, so the refusal
has to land on the request itself."""
with pytest.raises(BaseException) as excinfo:
async with SDKClient(
_server(handshake_only_server), mode="2026-07-28"
) as client:
await client.list_tools()
err = _find_mcp_error(excinfo.value)
assert err is not None
assert err.code == mcp_types.UNSUPPORTED_PROTOCOL_VERSION
assert "initialize" in err.message
# ---------------------------------------------------------------------------
# Enforcement: pinned single version, and the unrestricted default
# ---------------------------------------------------------------------------
async def test_pinned_version_allows_exact_match():
mcp = FastMCP("pinned", protocol_versions=["2025-11-25"])
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
async with SDKClient(_server(mcp), mode="legacy") as client:
assert client.protocol_version == "2025-11-25"
result = await client.list_tools()
assert [t.name for t in result.tools] == ["add"]
def _initialize_request(version: str) -> mcp_types.InitializeRequest:
return mcp_types.InitializeRequest.model_validate(
{
"method": "initialize",
"params": {
"protocolVersion": version,
"capabilities": {},
"clientInfo": {"name": "test", "version": "1"},
},
}
)
@pytest.mark.parametrize("offered", ["2024-11-05", "2025-03-26", "2025-06-18"])
def test_pinned_version_refuses_older_handshake(offered):
"""The SDK client cannot pin a handshake-era version through `mode`, so the
older-handshake refusal is exercised at the enforcement hook."""
mcp = FastMCP("pinned", protocol_versions=["2025-11-25"])
with pytest.raises(MCPError) as excinfo:
enforce_handshake_protocol_version(mcp, _initialize_request(offered))
assert "2025-11-25" in excinfo.value.message
assert excinfo.value.code == mcp_types.UNSUPPORTED_PROTOCOL_VERSION
def test_pinned_version_accepts_matching_handshake():
mcp = FastMCP("pinned", protocol_versions=["2025-11-25"])
enforce_handshake_protocol_version(mcp, _initialize_request("2025-11-25"))
def test_unrestricted_server_never_refuses_handshake():
mcp = FastMCP("open")
enforce_handshake_protocol_version(mcp, _initialize_request("2024-11-05"))
async def test_fastmcp_client_default_reaches_handshake_only_server():
"""`fastmcp.Client` defaults to the handshake in memory, which a
handshake-only server serves directly."""
mcp = FastMCP("handshake-only", protocol_versions=HANDSHAKE_PROTOCOL_VERSIONS)
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
async with Client(mcp) as client:
assert [t.name for t in await client.list_tools()] == ["add"]
async def test_fastmcp_client_needs_modern_mode_for_modern_only_server():
"""The mirror case: a modern-only server refuses the default in-memory
handshake, and the refusal names the modern protocol."""
mcp = FastMCP("modern-only", protocol_versions=MODERN_PROTOCOL_VERSIONS)
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
with pytest.raises(MCPError, match="server/discover"):
async with Client(mcp) as client:
await client.list_tools()
async with Client(mcp, mode="auto") as client:
assert [t.name for t in await client.list_tools()] == ["add"]
@pytest.mark.parametrize("mode", ["legacy", "auto", "2026-07-28"])
async def test_default_allows_every_era(mode):
mcp = FastMCP("open")
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
async with SDKClient(_server(mcp), mode=mode) as client:
result = await client.list_tools()
assert [t.name for t in result.tools] == ["add"]
# ---------------------------------------------------------------------------
# Startup coherence check (warnings only, silent unless declared)
# ---------------------------------------------------------------------------
def _coherence_warnings(caplog) -> list[str]:
return [
r.getMessage()
for r in caplog.records
if r.name == _COHERENCE_LOGGER and r.levelno == logging.WARNING
]
async def _warnings_from_startup(mcp: FastMCP, caplog) -> list[str]:
with caplog.at_level(logging.WARNING, logger=_COHERENCE_LOGGER):
async with mcp._lifespan_manager():
pass
return _coherence_warnings(caplog)
async def test_guard_tool_without_declaration_is_silent(caplog):
"""Declaring nothing asserts nothing, so there is no contradiction to warn
about — the guard tool raises its own clear era error if reached."""
mcp = FastMCP("guardy")
@mcp.tool
def ask(x: int) -> str | mcp_types.InputRequiredResult:
return "ok"
assert await _warnings_from_startup(mcp, caplog) == []
async def test_guard_tool_under_handshake_only_warns(caplog):
mcp = FastMCP("guardy", protocol_versions=HANDSHAKE_PROTOCOL_VERSIONS)
@mcp.tool
def ask(x: int) -> str | mcp_types.InputRequiredResult:
return "ok"
warnings = await _warnings_from_startup(mcp, caplog)
assert any("ask" in w and "InputRequiredResult" in w for w in warnings)
async def test_guard_tool_under_modern_declaration_silent(caplog):
mcp = FastMCP("guardy", protocol_versions=MODERN_PROTOCOL_VERSIONS)
@mcp.tool
def ask(x: int) -> str | mcp_types.InputRequiredResult:
return "ok"
assert await _warnings_from_startup(mcp, caplog) == []
async def test_guard_tool_under_mixed_declaration_silent(caplog):
"""A declaration that still includes a modern version can carry guard tools."""
mcp = FastMCP("guardy", protocol_versions=["2025-11-25", "2026-07-28"])
@mcp.tool
def ask(x: int) -> str | mcp_types.InputRequiredResult:
return "ok"
assert await _warnings_from_startup(mcp, caplog) == []
async def _sampling_handler(messages, params, context):
return "x"
async def test_fallback_sampling_under_modern_declaration_warns(caplog):
mcp = FastMCP(
"samp",
protocol_versions=MODERN_PROTOCOL_VERSIONS,
sampling_handler=_sampling_handler,
sampling_handler_behavior="fallback",
)
warnings = await _warnings_from_startup(mcp, caplog)
assert any("back-channel" in w and "fallback" in w for w in warnings)
async def test_fallback_sampling_without_declaration_is_silent(caplog):
mcp = FastMCP(
"samp",
sampling_handler=_sampling_handler,
sampling_handler_behavior="fallback",
)
assert await _warnings_from_startup(mcp, caplog) == []
async def test_always_sampling_under_modern_declaration_silent(caplog):
mcp = FastMCP(
"samp",
protocol_versions=MODERN_PROTOCOL_VERSIONS,
sampling_handler=_sampling_handler,
sampling_handler_behavior="always",
)
assert await _warnings_from_startup(mcp, caplog) == []
async def test_fallback_sampling_under_handshake_only_silent(caplog):
mcp = FastMCP(
"samp",
protocol_versions=HANDSHAKE_PROTOCOL_VERSIONS,
sampling_handler=_sampling_handler,
sampling_handler_behavior="fallback",
)
assert await _warnings_from_startup(mcp, caplog) == []
async def test_plain_server_is_coherent(caplog):
mcp = FastMCP("clean", protocol_versions=MODERN_PROTOCOL_VERSIONS)
@mcp.tool
def plain(a: int) -> int:
return a
assert await _warnings_from_startup(mcp, caplog) == []
async def test_guard_tool_reaches_modern_client():
"""A guard tool served under a modern declaration works end-to-end."""
mcp = FastMCP("guarded", protocol_versions=MODERN_PROTOCOL_VERSIONS)
@mcp.tool
async def confirm(ctx: Context) -> str | mcp_types.InputRequiredResult:
return "confirmed"
async with SDKClient(_server(mcp), mode="auto") as client:
result = await client.call_tool("confirm", {})
assert result.is_error is False