mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-24 06:24:18 +02:00
6fce4e538fEngine modules (keys, context snapshot, docket lifespan, worker CLI, client handles) move intact; SEP-1686 wire modules park in _legacy_wire for adaptation to SEP-2663. Core keeps task=True declaration on tools only and raises at serve time until the tasks extension is registered. Co-Authored-By: Claude <noreply@anthropic.com>
706 lines
26 KiB
Python
706 lines
26 KiB
Python
"""Dual-era protocol matrix: one FastMCP server served over both MCP protocol eras.
|
|
|
|
FastMCP must serve the legacy (initialize-handshake, 2025-11-25) era and the
|
|
modern (server/discover, 2026-07-28) era from the same server object. The test
|
|
harness is the v2 SDK's own first-class client, ``mcp.client.Client``, which
|
|
resolves an in-process ``Server`` directly:
|
|
|
|
* ``mode='legacy'`` forces the initialize handshake (2025-11-25 in-memory).
|
|
* ``mode='auto'`` probes ``server/discover`` and negotiates 2026-07-28.
|
|
* ``mode='2026-07-28'`` pins the modern version and adopts a synthesized
|
|
``DiscoverResult`` (no probe).
|
|
|
|
A FastMCP server exposes its lowlevel ``Server`` as ``fastmcp_server._mcp_server``;
|
|
that is what we hand to the SDK client, mirroring how
|
|
``mcp.client._memory.InMemoryTransport`` unwraps servers.
|
|
|
|
Several cells characterize behavior that is a verified SDK-era contract rather
|
|
than a FastMCP choice; those are flagged inline and cross-referenced to the
|
|
migration feedback dossier (``<scratchpad>/specs/sdk-feedback.md``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import mcp_types as types
|
|
import pytest
|
|
from mcp.client import Client as SDKClient
|
|
from mcp.client.session import ClientRequestContext
|
|
from mcp.server import Server as LowLevelServer
|
|
from mcp.shared.exceptions import MCPError
|
|
from pydantic import FileUrl
|
|
|
|
from fastmcp import Client as FastMCPClient
|
|
from fastmcp import Context, FastMCP, settings
|
|
from fastmcp.exceptions import PromptError, ResourceError
|
|
from fastmcp.server.elicitation import AcceptedElicitation
|
|
from fastmcp.server.middleware import Middleware
|
|
|
|
# Modes that reach the modern (2026-07-28) era via the SDK client.
|
|
MODERN_MODES = ["auto", "2026-07-28"]
|
|
# Both eras, for cells that must produce identical semantics on each.
|
|
ALL_MODES = ["legacy", *MODERN_MODES]
|
|
|
|
|
|
@pytest.fixture
|
|
def dual_era_server() -> FastMCP:
|
|
"""A single FastMCP server exercising every core MCP object type.
|
|
|
|
Deliberately minimal and side-effect free so the same instance can be
|
|
driven concurrently by legacy and modern clients within one test.
|
|
"""
|
|
mcp = FastMCP("dual-era")
|
|
|
|
@mcp.tool
|
|
def add(a: int, b: int) -> int:
|
|
"""Structured-output tool (returns a scalar wrapped as {"result": ...})."""
|
|
return a + b
|
|
|
|
@mcp.resource("data://config")
|
|
def config() -> dict:
|
|
return {"version": 1}
|
|
|
|
@mcp.resource("data://item/{item_id}")
|
|
def item(item_id: str) -> str:
|
|
return f"item-{item_id}"
|
|
|
|
@mcp.prompt
|
|
def summarize(topic: str) -> str:
|
|
return f"Summarize {topic}"
|
|
|
|
return mcp
|
|
|
|
|
|
def _server(mcp: FastMCP) -> LowLevelServer:
|
|
"""The lowlevel Server the SDK client connects to in-process."""
|
|
return mcp._mcp_server
|
|
|
|
|
|
def _texts(blocks) -> list[str]:
|
|
"""Text from CallToolResult.content blocks (TextContent)."""
|
|
return [b.text for b in blocks if isinstance(b, types.TextContent)]
|
|
|
|
|
|
def _resource_texts(blocks) -> list[str]:
|
|
"""Text from ReadResourceResult.contents blocks (TextResourceContents)."""
|
|
return [b.text for b in blocks if isinstance(b, types.TextResourceContents)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Core operations produce identical semantics on BOTH eras
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_list_tools_both_eras(dual_era_server, mode):
|
|
async with SDKClient(_server(dual_era_server), mode=mode) as client:
|
|
result = await client.list_tools()
|
|
assert [t.name for t in result.tools] == ["add"]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_call_tool_structured_output_both_eras(dual_era_server, mode):
|
|
async with SDKClient(_server(dual_era_server), mode=mode) as client:
|
|
result = await client.call_tool("add", {"a": 2, "b": 3})
|
|
assert result.is_error is False
|
|
assert result.structured_content == {"result": 5}
|
|
assert _texts(result.content) == ["5"]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_list_resources_both_eras(dual_era_server, mode):
|
|
async with SDKClient(_server(dual_era_server), mode=mode) as client:
|
|
result = await client.list_resources()
|
|
assert [str(r.uri) for r in result.resources] == ["data://config"]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_read_resource_both_eras(dual_era_server, mode):
|
|
async with SDKClient(_server(dual_era_server), mode=mode) as client:
|
|
result = await client.read_resource("data://config")
|
|
assert _resource_texts(result.contents) == ['{"version": 1}']
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_read_resource_template_both_eras(dual_era_server, mode):
|
|
async with SDKClient(_server(dual_era_server), mode=mode) as client:
|
|
result = await client.read_resource("data://item/42")
|
|
assert _resource_texts(result.contents) == ["item-42"]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_list_prompts_both_eras(dual_era_server, mode):
|
|
async with SDKClient(_server(dual_era_server), mode=mode) as client:
|
|
result = await client.list_prompts()
|
|
assert [p.name for p in result.prompts] == ["summarize"]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_get_prompt_both_eras(dual_era_server, mode):
|
|
async with SDKClient(_server(dual_era_server), mode=mode) as client:
|
|
result = await client.get_prompt("summarize", {"topic": "cats"})
|
|
rendered = [
|
|
m.content.text
|
|
for m in result.messages
|
|
if isinstance(m.content, types.TextContent)
|
|
]
|
|
assert rendered == ["Summarize cats"]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_complete_parity_both_eras(dual_era_server, mode):
|
|
"""FastMCP registers no completion handler, so `completion/complete` is
|
|
method-not-found. The point of this cell is parity: the *same* -32601
|
|
surfaces on both eras (2026 did not change the unsupported-method contract).
|
|
"""
|
|
async with SDKClient(_server(dual_era_server), mode=mode) as client:
|
|
with pytest.raises(MCPError) as excinfo:
|
|
await client.complete(
|
|
types.PromptReference(name="summarize"),
|
|
{"name": "topic", "value": "c"},
|
|
)
|
|
assert excinfo.value.code == types.METHOD_NOT_FOUND
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. Discovery / identity: which negotiation path each mode takes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def test_legacy_uses_initialize_handshake(dual_era_server):
|
|
"""Legacy mode runs the initialize handshake and reports a handshake-era
|
|
protocol version with server_info carried in the InitializeResult.
|
|
"""
|
|
async with SDKClient(_server(dual_era_server), mode="legacy") as client:
|
|
assert client.protocol_version == "2025-11-25"
|
|
assert client.server_info.name == "dual-era"
|
|
|
|
|
|
async def test_auto_negotiates_modern_via_discover(dual_era_server):
|
|
"""`mode='auto'` probes server/discover and adopts 2026-07-28, populating
|
|
server_info/capabilities from the DiscoverResult.
|
|
"""
|
|
async with SDKClient(_server(dual_era_server), mode="auto") as client:
|
|
assert client.protocol_version == "2026-07-28"
|
|
# server/discover carries identity, unlike the synthesized pin below.
|
|
assert client.server_info.name == "dual-era"
|
|
assert client.server_capabilities is not None
|
|
|
|
|
|
async def test_pinned_modern_adopts_without_probe(dual_era_server):
|
|
"""Pinning `mode='2026-07-28'` adopts the version directly. With no
|
|
`prior_discover`, the SDK synthesizes a minimal DiscoverResult, so
|
|
server_info is empty even though the protocol version is modern.
|
|
|
|
Characterization of the SDK's synthesize-discover path (mcp.client.client
|
|
`_synthesize_discover`): a pin without prior_discover trades identity for
|
|
skipping the probe round-trip.
|
|
"""
|
|
async with SDKClient(_server(dual_era_server), mode="2026-07-28") as client:
|
|
assert client.protocol_version == "2026-07-28"
|
|
assert client.server_info.name == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Push-feature degradation on 2026 vs. working callbacks on legacy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def push_server() -> FastMCP:
|
|
mcp = FastMCP("push")
|
|
|
|
@mcp.tool
|
|
async def do_elicit(ctx: Context) -> str:
|
|
result = await ctx.elicit("pick a value", response_type=int)
|
|
assert isinstance(result, AcceptedElicitation)
|
|
return f"elicited {result.data}"
|
|
|
|
@mcp.tool
|
|
async def do_sample(ctx: Context) -> str:
|
|
result = await ctx.sample("hello")
|
|
return f"sampled {result.text}"
|
|
|
|
@mcp.tool
|
|
async def do_list_roots(ctx: Context) -> str:
|
|
roots = await ctx.list_roots()
|
|
return f"roots {[str(r.uri) for r in roots]}"
|
|
|
|
@mcp.tool
|
|
async def do_log(ctx: Context) -> str:
|
|
await ctx.info("a log line")
|
|
return "logged"
|
|
|
|
return mcp
|
|
|
|
|
|
async def _accept_elicit(
|
|
context: ClientRequestContext, params: types.ElicitRequestParams
|
|
) -> types.ElicitResult:
|
|
return types.ElicitResult(action="accept", content={"value": 7})
|
|
|
|
|
|
async def _sampling_cb(
|
|
context: ClientRequestContext, params: types.CreateMessageRequestParams
|
|
) -> types.CreateMessageResult:
|
|
return types.CreateMessageResult(
|
|
role="assistant",
|
|
content=types.TextContent(type="text", text="sampled-text"),
|
|
model="test-model",
|
|
)
|
|
|
|
|
|
async def _roots_cb(context: ClientRequestContext) -> types.ListRootsResult:
|
|
return types.ListRootsResult(
|
|
roots=[types.Root(uri=FileUrl("file:///tmp"), name="tmp")]
|
|
)
|
|
|
|
|
|
async def test_elicit_works_on_legacy(push_server):
|
|
async with SDKClient(
|
|
_server(push_server), mode="legacy", elicitation_callback=_accept_elicit
|
|
) as client:
|
|
result = await client.call_tool("do_elicit", {})
|
|
assert result.is_error is False
|
|
assert _texts(result.content) == ["elicited 7"]
|
|
|
|
|
|
async def test_sample_works_on_legacy(push_server):
|
|
async with SDKClient(
|
|
_server(push_server), mode="legacy", sampling_callback=_sampling_cb
|
|
) as client:
|
|
result = await client.call_tool("do_sample", {})
|
|
assert result.is_error is False
|
|
assert _texts(result.content) == ["sampled sampled-text"]
|
|
|
|
|
|
async def test_list_roots_works_on_legacy(push_server):
|
|
async with SDKClient(
|
|
_server(push_server), mode="legacy", list_roots_callback=_roots_cb
|
|
) as client:
|
|
result = await client.call_tool("do_list_roots", {})
|
|
assert result.is_error is False
|
|
assert _texts(result.content) == ["roots ['file:///tmp']"]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", MODERN_MODES)
|
|
@pytest.mark.parametrize("tool", ["do_elicit", "do_sample", "do_list_roots"])
|
|
async def test_push_features_degrade_on_modern(push_server, mode, tool):
|
|
"""Server-initiated requests (elicitation/sampling/roots) are removed at
|
|
2026-07-28 (SEP-2577), so a tool that uses them must degrade to a surfaced
|
|
error rather than hang or crash the connection. This asserts the
|
|
degradation happens and reaches the caller as an isError result.
|
|
"""
|
|
async with SDKClient(
|
|
_server(push_server),
|
|
mode=mode,
|
|
elicitation_callback=_accept_elicit,
|
|
sampling_callback=_sampling_cb,
|
|
list_roots_callback=_roots_cb,
|
|
) as client:
|
|
result = await client.call_tool(tool, {})
|
|
assert result.is_error is True
|
|
# A subsequent normal call still works: the connection survived the
|
|
# per-request failure rather than tearing down the whole session.
|
|
log_result = await client.call_tool("do_log", {})
|
|
assert log_result.is_error is False
|
|
|
|
|
|
async def test_list_roots_degradation_message_is_clear_on_modern(push_server):
|
|
"""`ctx.list_roots()` sends with no related_request_id, so the SDK selects
|
|
the connection's no-back-channel outbound and raises the self-explanatory
|
|
NoBackChannelError. This is the *good* degradation message and we assert it.
|
|
"""
|
|
async with SDKClient(_server(push_server), mode="2026-07-28") as client:
|
|
result = await client.call_tool("do_list_roots", {})
|
|
assert result.is_error is True
|
|
message = " ".join(_texts(result.content)).lower()
|
|
assert "back-channel" in message and "server-initiated" in message
|
|
|
|
|
|
@pytest.mark.parametrize("tool", ["do_elicit", "do_sample"])
|
|
async def test_elicit_sample_degradation_message_is_clear_on_modern(push_server, tool):
|
|
"""FastMCP era-gates elicit/sample: on a 2026-07-28 connection they raise a
|
|
clear, era-aware error before hitting the wire, instead of the SDK's opaque
|
|
'Method not found' (sdk-feedback.md #10). Both messages name the removed
|
|
server-initiated capability so the caller knows why the request degraded.
|
|
"""
|
|
async with SDKClient(
|
|
_server(push_server),
|
|
mode="2026-07-28",
|
|
elicitation_callback=_accept_elicit,
|
|
sampling_callback=_sampling_cb,
|
|
) as client:
|
|
result = await client.call_tool(tool, {})
|
|
assert result.is_error is True
|
|
message = " ".join(_texts(result.content)).lower()
|
|
assert "server-initiated" in message
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3a-bis. Server-configured sampling handler answers WITHOUT the client
|
|
# back-channel, so ctx.sample()/ctx.sample_step() must keep working on modern
|
|
# connections. The era-gate only fires when nothing can serve the request.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _handler_server(behavior) -> FastMCP:
|
|
"""A server whose sampling is answered by a server-side handler."""
|
|
|
|
def sampling_handler(messages, params, ctx) -> str:
|
|
return "handler-answer"
|
|
|
|
mcp = FastMCP("handler", sampling_handler=sampling_handler)
|
|
if behavior is not None:
|
|
mcp.sampling_handler_behavior = behavior
|
|
|
|
@mcp.tool
|
|
async def do_sample(ctx: Context) -> str:
|
|
result = await ctx.sample("hello")
|
|
return f"sampled {result.text}"
|
|
|
|
@mcp.tool
|
|
async def do_sample_step(ctx: Context) -> str:
|
|
step = await ctx.sample_step("hello")
|
|
return f"stepped {step.text}"
|
|
|
|
return mcp
|
|
|
|
|
|
@pytest.mark.parametrize("mode", MODERN_MODES)
|
|
@pytest.mark.parametrize("behavior", ["always", "fallback"])
|
|
@pytest.mark.parametrize("method", ["do_sample", "do_sample_step"])
|
|
async def test_server_sampling_handler_works_on_modern(mode, behavior, method):
|
|
"""A server-side sampling handler answers entirely server-side, so it works
|
|
on modern (2026-07-28) connections regardless of behavior. The era-gate must
|
|
NOT block these — nothing touches the removed client back-channel. Crucially,
|
|
'fallback' must go straight to the handler (no bare client-attempt failure)."""
|
|
server = _handler_server(behavior)
|
|
async with SDKClient(_server(server), mode=mode) as client:
|
|
result = await client.call_tool(method, {})
|
|
assert result.is_error is False
|
|
assert "handler-answer" in " ".join(_texts(result.content))
|
|
|
|
|
|
@pytest.mark.parametrize("behavior", ["always", "fallback"])
|
|
@pytest.mark.parametrize("method", ["do_sample", "do_sample_step"])
|
|
async def test_server_sampling_handler_works_on_legacy(behavior, method):
|
|
"""Handshake-era behavior is unchanged: the server-side handler still answers
|
|
on legacy connections."""
|
|
server = _handler_server(behavior)
|
|
async with SDKClient(_server(server), mode="legacy") as client:
|
|
result = await client.call_tool(method, {})
|
|
assert result.is_error is False
|
|
assert "handler-answer" in " ".join(_texts(result.content))
|
|
|
|
|
|
@pytest.mark.parametrize("mode", MODERN_MODES)
|
|
@pytest.mark.parametrize("method", ["do_sample", "do_sample_step"])
|
|
async def test_sampling_without_handler_still_era_gated_on_modern(
|
|
push_server, mode, method
|
|
):
|
|
"""With no server-side handler configured, the request would hit the removed
|
|
client back-channel, so the clear era error still fires on modern."""
|
|
# push_server only defines do_sample; add a do_sample_step twin inline.
|
|
mcp = FastMCP("no-handler")
|
|
|
|
@mcp.tool
|
|
async def do_sample(ctx: Context) -> str:
|
|
result = await ctx.sample("hello")
|
|
return f"sampled {result.text}"
|
|
|
|
@mcp.tool
|
|
async def do_sample_step(ctx: Context) -> str:
|
|
step = await ctx.sample_step("hello")
|
|
return f"stepped {step.text}"
|
|
|
|
async with SDKClient(
|
|
_server(mcp), mode=mode, sampling_callback=_sampling_cb
|
|
) as client:
|
|
result = await client.call_tool(method, {})
|
|
assert result.is_error is True
|
|
assert "server-initiated" in " ".join(_texts(result.content)).lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3b. Sampling deprecation warning (SEP-2577): ctx.sample/ctx.sample_step warn
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def reset_sample_warn_flag():
|
|
"""Reset the process-wide warn-once flag so a warning can be observed."""
|
|
import fastmcp.server.context as context_module
|
|
|
|
original = set(context_module._sample_deprecation_warned)
|
|
context_module._sample_deprecation_warned.clear()
|
|
try:
|
|
yield
|
|
finally:
|
|
context_module._sample_deprecation_warned.clear()
|
|
context_module._sample_deprecation_warned.update(original)
|
|
|
|
|
|
@pytest.mark.parametrize("method", ["do_sample", "do_sample_step"])
|
|
async def test_sampling_emits_deprecation_warning(reset_sample_warn_flag, method):
|
|
"""`ctx.sample()` and `ctx.sample_step()` emit a FastMCPDeprecationWarning
|
|
naming SEP-2577 and the server-side-LLM migration path."""
|
|
from fastmcp.exceptions import FastMCPDeprecationWarning
|
|
|
|
mcp = FastMCP("warn")
|
|
|
|
@mcp.tool
|
|
async def do_sample(ctx: Context) -> str:
|
|
await ctx.sample("hello")
|
|
return "ok"
|
|
|
|
@mcp.tool
|
|
async def do_sample_step(ctx: Context) -> str:
|
|
await ctx.sample_step("hello")
|
|
return "ok"
|
|
|
|
with pytest.warns(FastMCPDeprecationWarning, match="SEP-2577"):
|
|
async with SDKClient(
|
|
_server(mcp), mode="legacy", sampling_callback=_sampling_cb
|
|
) as client:
|
|
await client.call_tool(method, {})
|
|
|
|
|
|
async def test_sampling_deprecation_warning_fires_once_per_process(
|
|
reset_sample_warn_flag,
|
|
):
|
|
"""The deprecation warning is warn-once: a second sample call in the same
|
|
process does not re-warn."""
|
|
from fastmcp.exceptions import FastMCPDeprecationWarning
|
|
|
|
mcp = FastMCP("warn-once")
|
|
|
|
@mcp.tool
|
|
async def do_sample(ctx: Context) -> str:
|
|
await ctx.sample("hello")
|
|
return "ok"
|
|
|
|
with pytest.warns(FastMCPDeprecationWarning):
|
|
async with SDKClient(
|
|
_server(mcp), mode="legacy", sampling_callback=_sampling_cb
|
|
) as client:
|
|
await client.call_tool("do_sample", {})
|
|
|
|
import warnings as _warnings
|
|
|
|
with _warnings.catch_warnings():
|
|
_warnings.simplefilter("error", FastMCPDeprecationWarning)
|
|
async with SDKClient(
|
|
_server(mcp), mode="legacy", sampling_callback=_sampling_cb
|
|
) as client:
|
|
result = await client.call_tool("do_sample", {})
|
|
assert result.is_error is False
|
|
|
|
|
|
async def test_sampling_deprecation_warning_suppressible_via_settings(
|
|
reset_sample_warn_flag, monkeypatch
|
|
):
|
|
"""Setting `deprecation_warnings=False` suppresses the sampling warning,
|
|
matching the house pattern for every other FastMCP deprecation."""
|
|
import warnings as _warnings
|
|
|
|
from fastmcp.exceptions import FastMCPDeprecationWarning
|
|
|
|
monkeypatch.setattr(settings, "deprecation_warnings", False)
|
|
|
|
mcp = FastMCP("no-warn")
|
|
|
|
@mcp.tool
|
|
async def do_sample(ctx: Context) -> str:
|
|
await ctx.sample("hello")
|
|
return "ok"
|
|
|
|
with _warnings.catch_warnings():
|
|
_warnings.simplefilter("error", FastMCPDeprecationWarning)
|
|
async with SDKClient(
|
|
_server(mcp), mode="legacy", sampling_callback=_sampling_cb
|
|
) as client:
|
|
result = await client.call_tool("do_sample", {})
|
|
assert result.is_error is False
|
|
|
|
|
|
@pytest.mark.parametrize("mode", MODERN_MODES)
|
|
async def test_logging_notification_still_flows_on_modern(push_server, mode):
|
|
"""`ctx.info` is a server->client *notification*, not a request. Unlike the
|
|
removed server-initiated requests, notifications still flow over the modern
|
|
direct-dispatcher path, so the tool completes successfully.
|
|
|
|
Characterization: current behavior is silent success (the log is emitted,
|
|
the tool returns normally); we assert that rather than an error.
|
|
"""
|
|
async with SDKClient(_server(push_server), mode=mode) as client:
|
|
result = await client.call_tool("do_log", {})
|
|
assert result.is_error is False
|
|
assert _texts(result.content) == ["logged"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. Sessionless safety: session-id-keyed paths must not crash on 2026 in-memory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def sessionless_server() -> FastMCP:
|
|
mcp = FastMCP("sessionless")
|
|
|
|
@mcp.tool
|
|
async def read_session_id(ctx: Context) -> str:
|
|
# In-memory/HTTP-less connections have no HTTP session id; FastMCP must
|
|
# synthesize a stable one rather than raise.
|
|
return ctx.session_id
|
|
|
|
return mcp
|
|
|
|
|
|
@pytest.mark.parametrize("mode", MODERN_MODES)
|
|
async def test_session_id_access_does_not_crash_on_modern(sessionless_server, mode):
|
|
async with SDKClient(_server(sessionless_server), mode=mode) as client:
|
|
result = await client.call_tool("read_session_id", {})
|
|
assert result.is_error is False
|
|
# A non-empty synthesized id was returned.
|
|
assert _texts(result.content)[0]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", MODERN_MODES)
|
|
async def test_set_logging_level_does_not_crash_on_modern(sessionless_server, mode):
|
|
"""logging/setLevel is a session-id-keyed, deprecated-at-2026 operation.
|
|
On a sessionless modern in-memory connection it must degrade cleanly (either
|
|
succeed as a no-op or raise a surfaced MCPError) rather than crash the
|
|
connection. Characterization: capture whichever the current contract is.
|
|
"""
|
|
async with SDKClient(_server(sessionless_server), mode=mode) as client:
|
|
outcome: str
|
|
try:
|
|
await client.set_logging_level("debug") # ty: ignore[deprecated]
|
|
outcome = "ok"
|
|
except MCPError:
|
|
outcome = "mcperror"
|
|
# Either way the connection is still usable afterward.
|
|
result = await client.call_tool("read_session_id", {})
|
|
assert result.is_error is False
|
|
assert outcome in {"ok", "mcperror"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. FastMCP middleware runs on both eras
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _CallToolCounter(Middleware):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.count = 0
|
|
|
|
async def on_call_tool(self, context, call_next):
|
|
self.count += 1
|
|
return await call_next(context)
|
|
|
|
|
|
async def test_middleware_runs_on_both_eras():
|
|
counter = _CallToolCounter()
|
|
mcp = FastMCP("mw")
|
|
mcp.add_middleware(counter)
|
|
|
|
@mcp.tool
|
|
def ping() -> str:
|
|
return "pong"
|
|
|
|
for mode in ("legacy", "2026-07-28"):
|
|
async with SDKClient(mcp._mcp_server, mode=mode) as client:
|
|
result = await client.call_tool("ping", {})
|
|
assert result.is_error is False
|
|
assert _texts(result.content) == ["pong"]
|
|
|
|
# One invocation observed from each era.
|
|
assert counter.count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Resource / prompt handler errors must survive both eras
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def erroring_server() -> FastMCP:
|
|
"""A server whose resource and prompt handlers raise FastMCP errors."""
|
|
mcp = FastMCP("erroring")
|
|
|
|
@mcp.resource("data://boom")
|
|
def boom() -> str:
|
|
raise ResourceError("resource detail marker")
|
|
|
|
@mcp.resource("data://items/{item_id}")
|
|
def item(item_id: int) -> str:
|
|
return f"item {item_id}"
|
|
|
|
@mcp.prompt
|
|
def explode() -> str:
|
|
raise PromptError("prompt detail marker")
|
|
|
|
return mcp
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_resource_error_message_reaches_client(
|
|
erroring_server: FastMCP, mode: str
|
|
) -> None:
|
|
"""A ResourceError's message must reach the wire on every era.
|
|
|
|
The modern runner masks any handler exception that is not an MCPError or a
|
|
ValidationError as a generic "Internal server error", so a ResourceError
|
|
that escapes the handler becomes indistinguishable from a server bug.
|
|
"""
|
|
async with FastMCPClient(erroring_server, mode=mode) as client:
|
|
with pytest.raises(MCPError) as exc_info:
|
|
await client.read_resource("data://boom")
|
|
|
|
assert "resource detail marker" in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_prompt_error_message_reaches_client(
|
|
erroring_server: FastMCP, mode: str
|
|
) -> None:
|
|
"""A PromptError's message must reach the wire on every era."""
|
|
async with FastMCPClient(erroring_server, mode=mode) as client:
|
|
with pytest.raises(MCPError) as exc_info:
|
|
await client.get_prompt("explode")
|
|
|
|
assert "prompt detail marker" in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_resource_template_conversion_error_reaches_client(
|
|
erroring_server: FastMCP, mode: str
|
|
) -> None:
|
|
"""A bad template argument is a client-input error, not a server fault.
|
|
|
|
This is the path that originally exposed the masking: converting
|
|
``item_id`` to an int fails, and the resulting error must name the problem
|
|
rather than surface as a generic internal error.
|
|
"""
|
|
async with FastMCPClient(erroring_server, mode=mode) as client:
|
|
with pytest.raises(MCPError) as exc_info:
|
|
await client.read_resource("data://items/not-an-int")
|
|
|
|
assert "Internal server error" not in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.parametrize("mode", ALL_MODES)
|
|
async def test_resource_error_masked_when_masking_enabled(mode: str) -> None:
|
|
"""Masking still applies: resources leak no more than tools already do."""
|
|
mcp = FastMCP("masked", mask_error_details=True)
|
|
|
|
@mcp.resource("data://boom")
|
|
def boom() -> str:
|
|
raise ValueError("secret internal detail")
|
|
|
|
async with FastMCPClient(mcp, mode=mode) as client:
|
|
with pytest.raises(MCPError) as exc_info:
|
|
await client.read_resource("data://boom")
|
|
|
|
assert "secret internal detail" not in str(exc_info.value)
|