modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Align with spec #3002: optional clientInfo, serverInfo in result _meta (#3143)

Browse source
This commit is contained in:
Max 2026-07-23 12:00:36 +01:00 • committed by GitHub
commit 837ef904f8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 1757 additions and 567 deletions
Download patch file Download diff file Expand all files Collapse all files

13
tests/interaction/_requirements.py
View file

@ -354,8 +354,8 @@ REQUIREMENTS: dict[str, Requirement] = {
"lifecycle:stateless:request-envelope": Requirement(
source=f"{SPEC_2026_BASE_URL}/basic/lifecycle#stateless-operation",
behavior=(
"At protocol_version 2026-07-28, every request carries io.modelcontextprotocol/protocolVersion, "
"/clientInfo, and /clientCapabilities in params._meta; no initialize handshake occurs."
"At protocol_version 2026-07-28, every request carries io.modelcontextprotocol/protocolVersion "
"and /clientCapabilities in params._meta (/clientInfo is optional); no initialize handshake occurs."
),
added_in="2026-07-28",
),
@ -408,7 +408,8 @@ REQUIREMENTS: dict[str, Requirement] = {
source=f"{SPEC_2026_BASE_URL}/basic/lifecycle#discover",
behavior=(
"Calling discover() sends server/discover with no params and returns a typed DiscoverResult "
"carrying protocolVersion, capabilities, serverInfo and the cache hint fields."
"carrying supportedVersions, capabilities and the cache hint fields; the server's identity "
"travels as the io.modelcontextprotocol/serverInfo stamp in the result _meta."
),
added_in="2026-07-28",
),
@ -3261,8 +3262,10 @@ REQUIREMENTS: dict[str, Requirement] = {
"hosting:http:modern:discover-response-shape": Requirement(
source=f"{SPEC_2026_BASE_URL}/basic/index",
behavior=(
"A 2026-07-28 server/discover response carries supportedVersions, capabilities, and "
"serverInfo, with supportedVersions naming the modern protocol revisions the server accepts."
"A 2026-07-28 server/discover response carries supportedVersions and capabilities in the "
"result body, with supportedVersions naming the modern protocol revisions the server "
"accepts; serverInfo is not a body field and travels as the io.modelcontextprotocol/serverInfo "
"result _meta stamp."
),
added_in="2026-07-28",
transports=("streamable-http",),

49
tests/interaction/conftest.py
View file

@ -6,10 +6,18 @@ the (transport, spec_version) cells via :func:`compute_cells`, applying arm excl
bounds, and known-failure xfails declaratively.
"""
from functools import partial
from contextlib import AbstractAsyncContextManager
from typing import Any
import pytest
from mcp_types import SERVER_INFO_META_KEY
from mcp_types.version import MODERN_PROTOCOL_VERSIONS
from mcp.client.client import Client
from mcp.server import Server
from mcp.server.mcpserver import MCPServer
from tests._stamp import R, Unstamp
from tests._stamp import unstamped as _strip_required_stamp
from tests.interaction._connect import (
Connect,
connect_in_memory,
@ -35,8 +43,23 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
metafunc.parametrize("connect", compute_cells(requirements), indirect=True)
class CellConnect:
"""The cell's connection factory, also naming the cell's `spec_version`.
Callable exactly like the `Connect` factories it wraps; the attribute lets
sibling fixtures (`unstamped`) key on the cell's era without re-deriving it.
"""
def __init__(self, factory: Connect, spec_version: str) -> None:
self._factory = factory
self.spec_version = spec_version
def __call__(self, server: Server | MCPServer, **kwargs: Any) -> AbstractAsyncContextManager[Client]:
return self._factory(server, spec_version=self.spec_version, **kwargs)
@pytest.fixture
def connect(request: pytest.FixtureRequest) -> Connect:
def connect(request: pytest.FixtureRequest) -> CellConnect:
"""The transport-parametrized connection factory: a test using it runs once per matrix cell.
Tests that are tied to one transport (the wire-recording tests, the bare-ClientSession tests,
@ -45,4 +68,24 @@ def connect(request: pytest.FixtureRequest) -> Connect:
transport, spec_version = request.param
assert isinstance(transport, str)
assert isinstance(spec_version, str)
return partial(_FACTORIES[transport], spec_version=spec_version)
return CellConnect(_FACTORIES[transport], spec_version)
@pytest.fixture
def unstamped(connect: CellConnect) -> Unstamp:
"""The cell's era-aware serverInfo-stamp normalizer, for full-result comparisons.
On a modern cell the stamp MUST be present (asserted) and is stripped so
one expected payload stays valid across eras; on a handshake-era cell the
result must not be stamped at all. Either direction failing is a runner
regression, so the same comparison line enforces both.
"""
if connect.spec_version in MODERN_PROTOCOL_VERSIONS:
return _strip_required_stamp
def _assert_never_stamped(result: R) -> R:
meta = result.meta
assert meta is None or SERVER_INFO_META_KEY not in meta, "handshake-era results are never stamped"
return result
return _assert_never_stamped

15
tests/interaction/lowlevel/test_cancellation.py
View file

@ -33,6 +33,7 @@ from mcp.client import ClientRequestContext, ClientSession
from mcp.server import Server, ServerRequestContext
from mcp.shared.memory import MessageStream, create_client_server_memory_streams
from mcp.shared.message import SessionMessage
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._helpers import IncomingMessage
from tests.interaction._requirements import requirement
@ -137,7 +138,7 @@ async def test_session_serves_requests_after_cancellation(connect: Connect) -> N
@requirement("protocol:cancel:unknown-id-ignored")
async def test_cancellation_for_unknown_request_is_ignored(connect: Connect) -> None:
async def test_cancellation_for_unknown_request_is_ignored(connect: Connect, unstamped: Unstamp) -> None:
"""A cancellation referencing a request id that is not in flight is ignored without error."""
async def list_tools(
@ -157,7 +158,7 @@ async def test_cancellation_for_unknown_request_is_ignored(connect: Connect) ->
)
result = await client.call_tool("echo", {})
assert result == snapshot(CallToolResult(content=[TextContent(text="unbothered")]))
assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="unbothered")]))
@requirement("protocol:cancel:server-to-client")
@ -350,7 +351,7 @@ async def test_timed_out_initialize_sends_no_cancellation() -> None:
@requirement("protocol:cancel:abort-signal")
async def test_abandoning_a_call_stops_the_server_handler(connect: Connect) -> None:
async def test_abandoning_a_call_stops_the_server_handler(connect: Connect, unstamped: Unstamp) -> None:
"""Cancelling the task that awaits a call cancels the request itself, not just the local wait:
the server-side handler is interrupted, and the session serves later requests normally.
@ -397,11 +398,11 @@ async def test_abandoning_a_call_stops_the_server_handler(connect: Connect) -> N
# dropped while the client is still open, so teardown never races its delivery.
await anyio.wait_all_tasks_blocked()
result = await client.call_tool("echo", {})
assert result == snapshot(CallToolResult(content=[TextContent(text="ok")]))
assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="ok")]))
@requirement("protocol:cancel:abort-scoped")
async def test_abandoning_one_call_leaves_a_concurrent_call_running(connect: Connect) -> None:
async def test_abandoning_one_call_leaves_a_concurrent_call_running(connect: Connect, unstamped: Unstamp) -> None:
"""Cancellation is scoped to the request it names: with two calls genuinely in flight,
abandoning the first interrupts only its handler and the second returns its result.
@ -464,4 +465,6 @@ async def test_abandoning_one_call_leaves_a_concurrent_call_running(connect: Con
# dropped while the client is still open, so teardown never races its delivery.
await anyio.wait_all_tasks_blocked()
assert results == snapshot([CallToolResult(content=[TextContent(text="survived")])])
assert [unstamped(result) for result in results] == snapshot(
[CallToolResult(content=[TextContent(text="survived")])]
)

13
tests/interaction/lowlevel/test_client_connect.py
View file

@ -25,6 +25,7 @@ from mcp_types import (
INVALID_REQUEST,
METHOD_NOT_FOUND,
PROTOCOL_VERSION_META_KEY,
SERVER_INFO_META_KEY,
UNSUPPORTED_PROTOCOL_VERSION,
DiscoverResult,
Implementation,
@ -123,13 +124,13 @@ async def test_prior_discover_populates_state_with_zero_connect_time_traffic() -
"""`Client(..., mode=<pin>, prior_discover=...)` sends nothing on entry and exposes the prior server_info.
Requirement `lifecycle:mode:prior-discover-zero-rtt` (sdk-defined): a previously-obtained
DiscoverResult is installed via `adopt()` so server_info and capabilities are available
immediately with zero round trips.
DiscoverResult is installed via `adopt()` so server_info (read from the result's `_meta`
serverInfo stamp) and capabilities are available immediately with zero round trips.
"""
prior = DiscoverResult(
supported_versions=[LATEST_MODERN_VERSION],
capabilities=ServerCapabilities(tools=ToolsCapability(list_changed=False)),
server_info=Implementation(name="cached-server", version="9.9.9"),
_meta={SERVER_INFO_META_KEY: {"name": "cached-server", "version": "9.9.9"}},
)
requests, on_request = _request_recorder()
@ -156,7 +157,8 @@ async def test_auto_mode_probes_server_discover_and_adopts_the_result() -> None:
Requirement `lifecycle:discover:basic` (spec basic/lifecycle#discover): the probe is a
single `server/discover` request whose result carries supported versions, capabilities,
server_info and the cache-hint fields, after which the session is modern-negotiated.
the cache-hint fields, and the `_meta` serverInfo stamp, after which the session is
modern-negotiated.
"""
requests, on_request = _request_recorder()
server = _tools_server("discoverable")
@ -167,6 +169,7 @@ async def test_auto_mode_probes_server_discover_and_adopts_the_result() -> None:
Client(streamable_http_client(f"{BASE_URL}/mcp", http_client=http), mode="auto") as client,
):
assert client.protocol_version == LATEST_MODERN_VERSION
assert client.server_info is not None
assert client.server_info.name == "discoverable"
await client.list_tools()
@ -198,7 +201,6 @@ async def test_auto_mode_retries_discover_once_on_unsupported_protocol_version()
return DiscoverResult(
supported_versions=list(MODERN_PROTOCOL_VERSIONS),
capabilities=ServerCapabilities(),
server_info=Implementation(name="picky", version="1.0.0"),
)
server = _tools_server("picky")
@ -309,6 +311,7 @@ async def test_auto_mode_falls_back_to_initialize_on_a_legacy_probe_rejection(
with anyio.fail_after(5):
async with Client(scripted_transport(), mode="auto") as client:
assert client.protocol_version == LATEST_HANDSHAKE_VERSION
assert client.server_info is not None
assert client.server_info.name == "legacy-only"
assert methods_seen == ["server/discover", "initialize", "notifications/initialized"]

15
tests/interaction/lowlevel/test_completion.py
View file

@ -15,6 +15,7 @@ from mcp_types import (
from mcp import MCPError
from mcp.server import Server, ServerRequestContext
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
@ -23,7 +24,7 @@ pytestmark = pytest.mark.anyio
@requirement("completion:prompt-arg")
@requirement("completion:result-shape")
async def test_complete_prompt_argument(connect: Connect) -> None:
async def test_complete_prompt_argument(connect: Connect, unstamped: Unstamp) -> None:
"""Completing a prompt argument delivers the ref, argument name, and current value to the handler.
The returned values are filtered by the argument's value, proving the value reached the handler.
@ -44,13 +45,13 @@ async def test_complete_prompt_argument(connect: Connect) -> None:
PromptReference(name="code_review"), argument={"name": "language", "value": "py"}
)
assert result == snapshot(
assert unstamped(result) == snapshot(
CompleteResult(completion=Completion(values=["python", "pytorch"], total=2, has_more=False))
)
@requirement("completion:resource-template-arg")
async def test_complete_resource_template_variable(connect: Connect) -> None:
async def test_complete_resource_template_variable(connect: Connect, unstamped: Unstamp) -> None:
"""Completing a URI template variable delivers the template URI and variable name to the handler."""
async def completion(ctx: ServerRequestContext, params: types.CompleteRequestParams) -> CompleteResult:
@ -67,11 +68,11 @@ async def test_complete_resource_template_variable(connect: Connect) -> None:
argument={"name": "owner", "value": "model"},
)
assert result == snapshot(CompleteResult(completion=Completion(values=["modelcontextprotocol"])))
assert unstamped(result) == snapshot(CompleteResult(completion=Completion(values=["modelcontextprotocol"])))
@requirement("completion:context-arguments")
async def test_complete_receives_context_arguments(connect: Connect) -> None:
async def test_complete_receives_context_arguments(connect: Connect, unstamped: Unstamp) -> None:
"""Previously-resolved arguments passed as completion context reach the handler.
The returned value is derived from the context, proving it arrived.
@ -92,7 +93,9 @@ async def test_complete_receives_context_arguments(connect: Connect) -> None:
context_arguments={"owner": "modelcontextprotocol"},
)
assert result == snapshot(CompleteResult(completion=Completion(values=["modelcontextprotocol/python-sdk"])))
assert unstamped(result) == snapshot(
CompleteResult(completion=Completion(values=["modelcontextprotocol/python-sdk"]))
)
@requirement("completion:error:invalid-ref")

11
tests/interaction/lowlevel/test_flows.py
View file

@ -32,6 +32,7 @@ from mcp import MCPError, UrlElicitationRequiredError
from mcp.client import ClientRequestContext
from mcp.server import Server, ServerRequestContext
from mcp.server.session import ServerSession
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._helpers import IncomingMessage
from tests.interaction._requirements import requirement
@ -53,7 +54,9 @@ def _list_tools(*names: str) -> ListToolsHandler:
@requirement("flow:tool-result:resource-link-follow")
async def test_a_resource_link_returned_by_a_tool_can_be_followed_with_read(connect: Connect) -> None:
async def test_a_resource_link_returned_by_a_tool_can_be_followed_with_read(
connect: Connect, unstamped: Unstamp
) -> None:
"""A tool returns a resource_link; reading that link's URI returns the referenced contents.
Steps: (1) call the tool, (2) extract the link from its content, (3) read_resource on the
@ -78,8 +81,10 @@ async def test_a_resource_link_returned_by_a_tool_can_be_followed_with_read(conn
assert isinstance(link, ResourceLink)
read = await client.read_resource(link.uri)
assert called == snapshot(CallToolResult(content=[ResourceLink(name="report", uri="file:///report.txt")]))
assert read == snapshot(
assert unstamped(called) == snapshot(
CallToolResult(content=[ResourceLink(name="report", uri="file:///report.txt")])
)
assert unstamped(read) == snapshot(
ReadResourceResult(contents=[TextResourceContents(uri="file:///report.txt", text="generated")])
)

5
tests/interaction/lowlevel/test_logging.py
View file

@ -11,6 +11,7 @@ from inline_snapshot import snapshot
from mcp_types import CallToolResult, EmptyResult, LoggingMessageNotificationParams, TextContent
from mcp.server import Server, ServerRequestContext
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
@ -46,7 +47,7 @@ async def test_set_logging_level_reaches_handler(connect: Connect) -> None:
@requirement("logging:message:fields")
@requirement("tools:call:logging-mid-execution")
async def test_log_messages_reach_logging_callback_in_order(connect: Connect) -> None:
async def test_log_messages_reach_logging_callback_in_order(connect: Connect, unstamped: Unstamp) -> None:
"""Log messages sent during a tool call arrive at the logging callback, in order, before the call returns.
The two messages pin the full notification shape: severity, optional logger name, and both
@ -83,7 +84,7 @@ async def test_log_messages_reach_logging_callback_in_order(connect: Connect) ->
async with connect(server, logging_callback=collect) as client:
result = await client.call_tool("chatty", {})
assert result == snapshot(CallToolResult(content=[TextContent(text="done")]))
assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="done")]))
assert received == snapshot(
[
LoggingMessageNotificationParams(level="info", logger="app.lifecycle", data="starting up"),

8
tests/interaction/lowlevel/test_meta.py
View file

@ -2,7 +2,8 @@
Meta is opaque pass-through data, so these tests assert identity against the value that was sent
rather than snapshotting a literal: the expected value and the sent value are the same variable,
which also proves the SDK injected nothing alongside it.
which also proves the SDK injected nothing alongside it beyond the 2026-era serverInfo stamp,
stripped via `unstamped` before comparison.
"""
import mcp_types as types
@ -10,6 +11,7 @@ import pytest
from mcp_types import CallToolResult, RequestParamsMeta, TextContent
from mcp.server import Server, ServerRequestContext
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
@ -42,7 +44,7 @@ async def test_request_meta_reaches_handler(connect: Connect) -> None:
@requirement("meta:result-to-client")
async def test_result_meta_reaches_client(connect: Connect) -> None:
async def test_result_meta_reaches_client(connect: Connect, unstamped: Unstamp) -> None:
"""The _meta object a handler attaches to its result is delivered to the client unchanged."""
result_meta = {"example.com/cost": 3}
@ -60,4 +62,4 @@ async def test_result_meta_reaches_client(connect: Connect) -> None:
async with connect(server) as client:
result = await client.call_tool("metered", {})
assert result == CallToolResult(content=[TextContent(text="done")], _meta=result_meta)
assert unstamped(result) == CallToolResult(content=[TextContent(text="done")], _meta=result_meta)

7
tests/interaction/lowlevel/test_pagination.py
View file

@ -22,6 +22,7 @@ from mcp_types import (
from mcp import MCPError
from mcp.server import Server, ServerRequestContext
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
@ -29,7 +30,7 @@ pytestmark = pytest.mark.anyio
@requirement("tools:list:pagination")
async def test_next_cursor_round_trips_through_the_client(connect: Connect) -> None:
async def test_next_cursor_round_trips_through_the_client(connect: Connect, unstamped: Unstamp) -> None:
"""The next_cursor a list handler returns reaches the client, and the cursor the client sends
back on the following call reaches the handler verbatim.
"""
@ -55,7 +56,9 @@ async def test_next_cursor_round_trips_through_the_client(connect: Connect) -> N
assert first_page.next_cursor == cursor
assert seen_cursors == [None, cursor]
assert [tool.name for tool in first_page.tools] == ["alpha"]
assert second_page == snapshot(ListToolsResult(tools=[Tool(name="beta", input_schema={"type": "object"})]))
assert unstamped(second_page) == snapshot(
ListToolsResult(tools=[Tool(name="beta", input_schema={"type": "object"})])
)
@requirement("pagination:exhaustion")

9
tests/interaction/lowlevel/test_progress.py
View file

@ -18,6 +18,7 @@ from mcp_types import CallToolResult, ProgressNotification, ProgressNotification
from mcp.server import Server, ServerRequestContext
from mcp.server.session import ServerSession
from mcp.shared.session import ProgressFnT
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._helpers import IncomingMessage
from tests.interaction._requirements import requirement
@ -27,7 +28,7 @@ pytestmark = pytest.mark.anyio
@requirement("protocol:progress:callback")
@requirement("tools:call:progress")
async def test_progress_during_tool_call_reaches_callback_in_order(connect: Connect) -> None:
async def test_progress_during_tool_call_reaches_callback_in_order(connect: Connect, unstamped: Unstamp) -> None:
"""Progress notifications emitted by a tool handler reach the caller's progress callback in order."""
received: list[tuple[float, float | None, str | None]] = []
@ -51,7 +52,7 @@ async def test_progress_during_tool_call_reaches_callback_in_order(connect: Conn
async with connect(server) as client:
result = await client.call_tool("download", {}, progress_callback=collect)
assert result == snapshot(CallToolResult(content=[TextContent(text="downloaded")]))
assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="downloaded")]))
assert received == snapshot([(1.0, 3.0, "first chunk"), (2.0, 3.0, "second chunk"), (3.0, 3.0, "done")])
@ -83,7 +84,7 @@ async def test_progress_token_visible_to_handler(connect: Connect) -> None:
@requirement("protocol:progress:no-token")
async def test_no_progress_callback_means_no_token(connect: Connect) -> None:
async def test_no_progress_callback_means_no_token(connect: Connect, unstamped: Unstamp) -> None:
"""Without a progress callback the request carries no progress token.
The low-level API has no way to report request-scoped progress without a token, so a handler
@ -105,7 +106,7 @@ async def test_no_progress_callback_means_no_token(connect: Connect) -> None:
async with connect(server) as client:
result = await client.call_tool("inspect", {})
assert result == snapshot(CallToolResult(content=[TextContent(text="None")]))
assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="None")]))
@requirement("protocol:progress:client-to-server")

21
tests/interaction/lowlevel/test_prompts.py
View file

@ -21,6 +21,7 @@ from mcp_types import (
from mcp import MCPError
from mcp.server import Server, ServerRequestContext
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
@ -28,7 +29,7 @@ pytestmark = pytest.mark.anyio
@requirement("prompts:list:basic")
async def test_list_prompts_returns_registered_prompts(connect: Connect) -> None:
async def test_list_prompts_returns_registered_prompts(connect: Connect, unstamped: Unstamp) -> None:
"""The prompts returned by the handler reach the client with their argument declarations intact."""
async def list_prompts(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListPromptsResult:
@ -52,7 +53,7 @@ async def test_list_prompts_returns_registered_prompts(connect: Connect) -> None
async with connect(server) as client:
result = await client.list_prompts()
assert result == snapshot(
assert unstamped(result) == snapshot(
ListPromptsResult(
prompts=[
Prompt(
@ -71,7 +72,7 @@ async def test_list_prompts_returns_registered_prompts(connect: Connect) -> None
@requirement("prompts:get:with-args")
async def test_get_prompt_substitutes_arguments(connect: Connect) -> None:
async def test_get_prompt_substitutes_arguments(connect: Connect, unstamped: Unstamp) -> None:
"""Arguments supplied by the client reach the prompt handler; the templated message comes back."""
async def get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestParams) -> GetPromptResult:
@ -87,7 +88,7 @@ async def test_get_prompt_substitutes_arguments(connect: Connect) -> None:
async with connect(server) as client:
result = await client.get_prompt("greet", {"name": "Ada"})
assert result == snapshot(
assert unstamped(result) == snapshot(
GetPromptResult(
description="A personalised greeting.",
messages=[PromptMessage(role="user", content=TextContent(text="Hello, Ada!"))],
@ -96,7 +97,7 @@ async def test_get_prompt_substitutes_arguments(connect: Connect) -> None:
@requirement("prompts:get:multi-message")
async def test_get_prompt_multiple_messages_preserve_roles_and_order(connect: Connect) -> None:
async def test_get_prompt_multiple_messages_preserve_roles_and_order(connect: Connect, unstamped: Unstamp) -> None:
"""A prompt returning a user/assistant conversation reaches the client with roles and order intact."""
async def get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestParams) -> GetPromptResult:
@ -114,7 +115,7 @@ async def test_get_prompt_multiple_messages_preserve_roles_and_order(connect: Co
async with connect(server) as client:
result = await client.get_prompt("geography_quiz")
assert result == snapshot(
assert unstamped(result) == snapshot(
GetPromptResult(
messages=[
PromptMessage(role="user", content=TextContent(text="What is the capital of France?")),
@ -126,7 +127,7 @@ async def test_get_prompt_multiple_messages_preserve_roles_and_order(connect: Co
@requirement("prompts:get:no-args")
async def test_get_prompt_without_arguments_returns_the_messages(connect: Connect) -> None:
async def test_get_prompt_without_arguments_returns_the_messages(connect: Connect, unstamped: Unstamp) -> None:
"""A prompt fetched with no arguments delivers None as the handler's arguments and returns its messages."""
async def get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestParams) -> GetPromptResult:
@ -139,7 +140,7 @@ async def test_get_prompt_without_arguments_returns_the_messages(connect: Connec
async with connect(server) as client:
result = await client.get_prompt("static")
assert result == snapshot(
assert unstamped(result) == snapshot(
GetPromptResult(messages=[PromptMessage(role="user", content=TextContent(text="Say hello."))])
)
@ -147,7 +148,7 @@ async def test_get_prompt_without_arguments_returns_the_messages(connect: Connec
@requirement("prompts:get:content:image")
@requirement("prompts:get:content:audio")
@requirement("prompts:get:content:embedded-resource")
async def test_get_prompt_with_non_text_content_round_trips(connect: Connect) -> None:
async def test_get_prompt_with_non_text_content_round_trips(connect: Connect, unstamped: Unstamp) -> None:
"""Prompt messages can carry image, audio, and embedded-resource content; all reach the client.
A single full-result snapshot proves all three content types round-trip: each block in the result
@ -175,7 +176,7 @@ async def test_get_prompt_with_non_text_content_round_trips(connect: Connect) ->
async with connect(server) as client:
result = await client.get_prompt("media", {})
assert result == snapshot(
assert unstamped(result) == snapshot(
GetPromptResult(
messages=[
PromptMessage(role="user", content=ImageContent(data="aW1n", mime_type="image/png")),

17
tests/interaction/lowlevel/test_resources.py
View file

@ -27,6 +27,7 @@ from mcp_types import (
from mcp import MCPError
from mcp.server import Server, ServerRequestContext
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._helpers import IncomingMessage
from tests.interaction._requirements import requirement
@ -36,7 +37,7 @@ pytestmark = pytest.mark.anyio
@requirement("resources:list:basic")
@requirement("resources:annotations")
async def test_list_resources_returns_registered_resources(connect: Connect) -> None:
async def test_list_resources_returns_registered_resources(connect: Connect, unstamped: Unstamp) -> None:
"""Listed resources reach the client with their URIs, names, and optional descriptive fields intact.
The fully-populated entry includes annotations, so the snapshot also proves they round-trip.
@ -71,7 +72,7 @@ async def test_list_resources_returns_registered_resources(connect: Connect) ->
async with connect(server) as client:
result = await client.list_resources()
assert result == snapshot(
assert unstamped(result) == snapshot(
ListResourcesResult(
resources=[
Resource(uri="memo://minimal", name="minimal"),
@ -93,7 +94,7 @@ async def test_list_resources_returns_registered_resources(connect: Connect) ->
@requirement("resources:read:text")
async def test_read_resource_text(connect: Connect) -> None:
async def test_read_resource_text(connect: Connect, unstamped: Unstamp) -> None:
"""Reading a text resource returns its contents with the URI, MIME type, and text supplied by the handler."""
async def read_resource(ctx: ServerRequestContext, params: types.ReadResourceRequestParams) -> ReadResourceResult:
@ -106,7 +107,7 @@ async def test_read_resource_text(connect: Connect) -> None:
async with connect(server) as client:
result = await client.read_resource("file:///greeting.txt")
assert result == snapshot(
assert unstamped(result) == snapshot(
ReadResourceResult(
contents=[TextResourceContents(uri="file:///greeting.txt", mime_type="text/plain", text="Hello, world!")]
)
@ -114,7 +115,7 @@ async def test_read_resource_text(connect: Connect) -> None:
@requirement("resources:read:blob")
async def test_read_resource_binary(connect: Connect) -> None:
async def test_read_resource_binary(connect: Connect, unstamped: Unstamp) -> None:
"""Reading a binary resource returns its contents base64-encoded in the blob field."""
async def read_resource(ctx: ServerRequestContext, params: types.ReadResourceRequestParams) -> ReadResourceResult:
@ -133,7 +134,7 @@ async def test_read_resource_binary(connect: Connect) -> None:
async with connect(server) as client:
result = await client.read_resource("file:///pixel.png")
assert result == snapshot(
assert unstamped(result) == snapshot(
ReadResourceResult(
contents=[BlobResourceContents(uri="file:///pixel.png", mime_type="image/png", blob="iVBORw==")]
)
@ -161,7 +162,7 @@ async def test_read_resource_unknown_uri_is_protocol_error(connect: Connect) ->
@requirement("resources:templates:list")
async def test_list_resource_templates_returns_registered_templates(connect: Connect) -> None:
async def test_list_resource_templates_returns_registered_templates(connect: Connect, unstamped: Unstamp) -> None:
"""Listed resource templates reach the client with their URI templates and descriptive fields intact."""
async def list_resource_templates(
@ -186,7 +187,7 @@ async def test_list_resource_templates_returns_registered_templates(connect: Con
async with connect(server) as client:
result = await client.list_resource_templates()
assert result == snapshot(
assert unstamped(result) == snapshot(
ListResourceTemplatesResult(
resource_templates=[
ResourceTemplate(uri_template="users://{user_id}", name="user"),

49
tests/interaction/lowlevel/test_tools.py
View file

@ -22,6 +22,7 @@ from mcp_types import (
from mcp import MCPError
from mcp.server import Server, ServerRequestContext
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
@ -29,7 +30,7 @@ pytestmark = pytest.mark.anyio
@requirement("tools:call:content:text")
async def test_call_tool_returns_text_content(connect: Connect) -> None:
async def test_call_tool_returns_text_content(connect: Connect, unstamped: Unstamp) -> None:
"""Arguments reach the tool handler; its content comes back as the call result."""
async def list_tools(
@ -49,11 +50,11 @@ async def test_call_tool_returns_text_content(connect: Connect) -> None:
async with connect(server) as client:
result = await client.call_tool("add", {"a": 2, "b": 3})
assert result == snapshot(CallToolResult(content=[TextContent(text="5")]))
assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="5")]))
@requirement("tools:call:is-error")
async def test_call_tool_execution_error_is_returned_as_result(connect: Connect) -> None:
async def test_call_tool_execution_error_is_returned_as_result(connect: Connect, unstamped: Unstamp) -> None:
"""A tool reporting its own failure with is_error=True reaches the client as a result, not an exception.
Tool execution errors are part of the result so the caller (typically a model) can see
@ -69,7 +70,7 @@ async def test_call_tool_execution_error_is_returned_as_result(connect: Connect)
async with connect(server) as client:
result = await client.call_tool("flux", {})
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="the flux capacitor is offline")], is_error=True)
)
@ -117,7 +118,7 @@ async def test_call_tool_uncaught_exception_becomes_error_response(connect: Conn
@requirement("tools:list:basic")
async def test_list_tools_returns_registered_tools(connect: Connect) -> None:
async def test_list_tools_returns_registered_tools(connect: Connect, unstamped: Unstamp) -> None:
"""The tools advertised by the server's list handler arrive at the client unchanged."""
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
@ -141,7 +142,7 @@ async def test_list_tools_returns_registered_tools(connect: Connect) -> None:
async with connect(server) as client:
result = await client.list_tools()
assert result == snapshot(
assert unstamped(result) == snapshot(
ListToolsResult(
tools=[
Tool(
@ -163,7 +164,7 @@ async def test_list_tools_returns_registered_tools(connect: Connect) -> None:
@requirement("tools:input-schema:preserve-additional-properties")
@requirement("tools:input-schema:preserve-defs")
@requirement("tools:input-schema:preserve-schema-dialect")
async def test_tools_list_preserves_arbitrary_input_schema_keywords(connect: Connect) -> None:
async def test_tools_list_preserves_arbitrary_input_schema_keywords(connect: Connect, unstamped: Unstamp) -> None:
"""A rich JSON Schema 2020-12 inputSchema reaches the client unchanged and the tool is callable.
The single identity assertion below proves all four pass-through behaviours at once: the same
@ -202,11 +203,11 @@ async def test_tools_list_preserves_arbitrary_input_schema_keywords(connect: Con
called = await client.call_tool("typed", {"count": 3, "options": {"verbose": True}})
assert listed.tools[0].input_schema == schema
assert called == snapshot(CallToolResult(content=[TextContent(text="ok")]))
assert unstamped(called) == snapshot(CallToolResult(content=[TextContent(text="ok")]))
@requirement("tools:list:metadata")
async def test_list_tools_optional_fields_round_trip(connect: Connect) -> None:
async def test_list_tools_optional_fields_round_trip(connect: Connect, unstamped: Unstamp) -> None:
"""Every optional Tool field the server supplies reaches the client unchanged."""
tool = Tool(
@ -228,7 +229,7 @@ async def test_list_tools_optional_fields_round_trip(connect: Connect) -> None:
async with connect(server) as client:
result = await client.list_tools()
assert result == snapshot(
assert unstamped(result) == snapshot(
ListToolsResult(
tools=[
Tool(
@ -251,7 +252,7 @@ async def test_list_tools_optional_fields_round_trip(connect: Connect) -> None:
@requirement("tools:call:content:audio")
@requirement("tools:call:content:resource-link")
@requirement("tools:call:content:embedded-resource")
async def test_call_tool_multiple_content_block_types(connect: Connect) -> None:
async def test_call_tool_multiple_content_block_types(connect: Connect, unstamped: Unstamp) -> None:
"""A tool result can mix every content block type; all of them arrive in order.
The payloads are tiny fixed base64 strings ("aW1n" is b"img", "YXVk" is b"aud") so the
@ -280,7 +281,7 @@ async def test_call_tool_multiple_content_block_types(connect: Connect) -> None:
async with connect(server) as client:
result = await client.call_tool("render", {})
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(
content=[
TextContent(text="all five content block types"),
@ -296,7 +297,7 @@ async def test_call_tool_multiple_content_block_types(connect: Connect) -> None:
@requirement("tools:call:structured-content")
async def test_call_tool_structured_content(connect: Connect) -> None:
async def test_call_tool_structured_content(connect: Connect, unstamped: Unstamp) -> None:
"""A tool result carrying structured content alongside content delivers both to the client."""
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
@ -311,11 +312,13 @@ async def test_call_tool_structured_content(connect: Connect) -> None:
async with connect(server) as client:
result = await client.call_tool("sum", {})
assert result == snapshot(CallToolResult(content=[TextContent(text="the sum is 5")], structured_content={"sum": 5}))
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="the sum is 5")], structured_content={"sum": 5})
)
@requirement("tools:call:concurrent")
async def test_concurrent_tool_calls_complete_independently(connect: Connect) -> None:
async def test_concurrent_tool_calls_complete_independently(connect: Connect, unstamped: Unstamp) -> None:
"""Two tool calls in flight at once run concurrently and each caller gets its own answer.
Both handlers are held on a shared event after signalling that they have started, and the test
@ -347,7 +350,7 @@ async def test_concurrent_tool_calls_complete_independently(connect: Connect) ->
async with anyio.create_task_group() as task_group: # pragma: no branch
async def call_and_record(tag: str) -> None:
results[tag] = await client.call_tool("echo", {"tag": tag})
results[tag] = unstamped(await client.call_tool("echo", {"tag": tag}))
task_group.start_soon(call_and_record, "first")
task_group.start_soon(call_and_record, "second")
@ -403,7 +406,7 @@ async def test_call_tool_structured_content_violating_output_schema_is_rejected_
@requirement("client:output-schema:skip-on-error")
async def test_is_error_result_bypasses_client_output_schema_validation(connect: Connect) -> None:
async def test_is_error_result_bypasses_client_output_schema_validation(connect: Connect, unstamped: Unstamp) -> None:
"""A tool result with isError true is returned as-is even when its structured content violates the schema.
The schema is cached up front so the client could validate, proving the bypass is specifically the
@ -437,7 +440,7 @@ async def test_is_error_result_bypasses_client_output_schema_validation(connect:
await client.list_tools()
result = await client.call_tool("forecast", {})
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="boom")], structured_content={"temperature": "warm"}, is_error=True)
)
@ -475,7 +478,9 @@ async def test_declared_output_schema_with_no_structured_content_is_rejected_by_
@requirement("client:output-schema:auto-list")
async def test_call_tool_populates_the_output_schema_cache_via_an_implicit_tools_list(connect: Connect) -> None:
async def test_call_tool_populates_the_output_schema_cache_via_an_implicit_tools_list(
connect: Connect, unstamped: Unstamp
) -> None:
"""Calling a tool whose schema is not cached issues exactly one implicit tools/list to populate it.
The first call_tool of an uncached tool triggers a tools/list the caller never asked for; the
@ -509,5 +514,7 @@ async def test_call_tool_populates_the_output_schema_cache_via_an_implicit_tools
second = await client.call_tool("forecast", {})
assert list_calls == ["called"]
assert first == snapshot(CallToolResult(content=[TextContent(text="21 C")], structured_content={"temperature": 21}))
assert second == first
assert unstamped(first) == snapshot(
CallToolResult(content=[TextContent(text="21 C")], structured_content={"temperature": 21})
)
assert unstamped(second) == first

13
tests/interaction/mcpserver/test_context.py
View file

@ -20,6 +20,7 @@ from mcp import MCPError
from mcp.client import ClientRequestContext
from mcp.server.elicitation import AcceptedElicitation
from mcp.server.mcpserver import Context, MCPServer
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._helpers import IncomingMessage
from tests.interaction._requirements import requirement
@ -68,7 +69,7 @@ async def test_context_logging_helpers_send_log_notifications(connect: Connect)
@requirement("mcpserver:context:progress")
async def test_context_report_progress_sends_progress_notifications(connect: Connect) -> None:
async def test_context_report_progress_sends_progress_notifications(connect: Connect, unstamped: Unstamp) -> None:
"""Context.report_progress sends progress notifications correlated to the calling request.
The caller's progress callback receives each report, in order, before the tool call returns.
@ -88,7 +89,7 @@ async def test_context_report_progress_sends_progress_notifications(connect: Con
async with connect(mcp) as client:
result = await client.call_tool("crunch", {}, progress_callback=on_progress)
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="crunched")], structured_content={"result": "crunched"})
)
assert received == snapshot([(1.0, 3.0, None), (2.0, 3.0, "halfway there")])
@ -123,7 +124,7 @@ async def test_context_exposes_request_id_and_client_info_to_a_tool(connect: Con
@requirement("mcpserver:context:logging")
@requirement("protocol:progress:no-token")
async def test_report_progress_without_a_progress_token_sends_nothing(connect: Connect) -> None:
async def test_report_progress_without_a_progress_token_sends_nothing(connect: Connect, unstamped: Unstamp) -> None:
"""When the caller supplied no progress callback, Context.report_progress is a silent no-op.
The tool also emits one log message as a sentinel: the message handler receives only that,
@ -145,7 +146,7 @@ async def test_report_progress_without_a_progress_token_sends_nothing(connect: C
async with connect(mcp, message_handler=collect) as client:
result = await client.call_tool("mill", {})
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="milled")], structured_content={"result": "milled"})
)
assert received == snapshot(
@ -207,7 +208,7 @@ async def test_context_elicit_returns_typed_result(connect: Connect) -> None:
@requirement("mcpserver:context:read-resource")
async def test_context_read_resource_reads_registered_resource(connect: Connect) -> None:
async def test_context_read_resource_reads_registered_resource(connect: Connect, unstamped: Unstamp) -> None:
"""Context.read_resource lets a tool read a resource registered on the same server.
The tool reports the MIME type and content it read, proving the resource function ran and its
@ -228,7 +229,7 @@ async def test_context_read_resource_reads_registered_resource(connect: Connect)
async with connect(mcp) as client:
result = await client.call_tool("show_config", {})
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(
content=[TextContent(text="text/plain: 'theme = dark'")],
structured_content={"result": "text/plain: 'theme = dark'"},

7
tests/interaction/mcpserver/test_extensions.py
View file

@ -15,6 +15,7 @@ from mcp.client import ClaimContext, ClientExtension, ResultClaim, advertise
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.extension import Extension
from mcp.server.mcpserver import Context, MCPServer, require_client_extension
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
@ -79,7 +80,9 @@ def _receipt_shop(issuer: Extension) -> MCPServer:
@requirement("extensions:client:claimed-result-resolved")
async def test_claimed_result_is_finished_by_the_owning_extensions_resolver(connect: Connect) -> None:
async def test_claimed_result_is_finished_by_the_owning_extensions_resolver(
connect: Connect, unstamped: Unstamp
) -> None:
"""The owning extension's claim resolver redeems the substituted `receipt` through
`ctx.session`, and `call_tool` returns the resolver's plain `CallToolResult`."""
received: list[ReceiptResult] = []
@ -92,7 +95,7 @@ async def test_claimed_result_is_finished_by_the_owning_extensions_resolver(conn
result = await client.call_tool("buy", {"item": "lamp"})
assert [claimed.receipt_token for claimed in received] == ["r-117"]
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="goods for r-117")], structured_content={"result": "goods for r-117"})
)

21
tests/interaction/mcpserver/test_prompts.py
View file

@ -14,6 +14,7 @@ from mcp_types import (
from mcp import MCPError
from mcp.server.mcpserver import MCPServer
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
@ -21,7 +22,7 @@ pytestmark = pytest.mark.anyio
@requirement("mcpserver:prompt:decorated")
async def test_list_prompts_derives_arguments_from_signature(connect: Connect) -> None:
async def test_list_prompts_derives_arguments_from_signature(connect: Connect, unstamped: Unstamp) -> None:
"""A decorated prompt is listed with arguments derived from the function signature.
Parameters without a default are required; the description comes from the docstring.
@ -36,7 +37,7 @@ async def test_list_prompts_derives_arguments_from_signature(connect: Connect) -
async with connect(mcp) as client:
result = await client.list_prompts()
assert result == snapshot(
assert unstamped(result) == snapshot(
ListPromptsResult(
prompts=[
Prompt(
@ -53,7 +54,7 @@ async def test_list_prompts_derives_arguments_from_signature(connect: Connect) -
@requirement("mcpserver:prompt:decorated")
async def test_get_prompt_renders_function_return(connect: Connect) -> None:
async def test_get_prompt_renders_function_return(connect: Connect, unstamped: Unstamp) -> None:
"""The decorated function's string return value is rendered as a single user message."""
mcp = MCPServer("prompter")
@ -65,7 +66,7 @@ async def test_get_prompt_renders_function_return(connect: Connect) -> None:
async with connect(mcp) as client:
result = await client.get_prompt("greet", {"name": "Ada"})
assert result == snapshot(
assert unstamped(result) == snapshot(
GetPromptResult(
description="A personalised greeting.",
messages=[PromptMessage(role="user", content=TextContent(text="Say hello to Ada."))],
@ -141,7 +142,9 @@ async def test_get_prompt_with_a_wrong_type_argument_is_rejected_before_the_func
@requirement("mcpserver:prompt:optional-args")
async def test_get_prompt_with_an_optional_argument_omitted_uses_the_default(connect: Connect) -> None:
async def test_get_prompt_with_an_optional_argument_omitted_uses_the_default(
connect: Connect, unstamped: Unstamp
) -> None:
"""A prompt rendered without one of its optional arguments uses that parameter's default value."""
mcp = MCPServer("prompter")
@ -153,7 +156,7 @@ async def test_get_prompt_with_an_optional_argument_omitted_uses_the_default(con
async with connect(mcp) as client:
result = await client.get_prompt("review", {"code": "x = 1"})
assert result == snapshot(
assert unstamped(result) == snapshot(
GetPromptResult(
description="Review a snippet of code against a style guide.",
messages=[PromptMessage(role="user", content=TextContent(text="Review x = 1 per pep8."))],
@ -162,7 +165,9 @@ async def test_get_prompt_with_an_optional_argument_omitted_uses_the_default(con
@requirement("mcpserver:prompt:duplicate-name")
async def test_registering_a_duplicate_prompt_name_warns_and_keeps_the_first(connect: Connect) -> None:
async def test_registering_a_duplicate_prompt_name_warns_and_keeps_the_first(
connect: Connect, unstamped: Unstamp
) -> None:
"""Registering a second prompt with an already-used name keeps the first registration.
The intended behaviour is rejection at registration time; MCPServer instead logs a warning
@ -187,7 +192,7 @@ async def test_registering_a_duplicate_prompt_name_warns_and_keeps_the_first(con
result = await client.get_prompt("greet")
assert [prompt.name for prompt in listed.prompts] == ["greet"]
assert result == snapshot(
assert unstamped(result) == snapshot(
GetPromptResult(
description="The first registration; this is the one that wins.",
messages=[PromptMessage(role="user", content=TextContent(text="first"))],

21
tests/interaction/mcpserver/test_resources.py
View file

@ -14,6 +14,7 @@ from mcp_types import (
from mcp import MCPError
from mcp.server.mcpserver import MCPServer
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
@ -21,7 +22,7 @@ pytestmark = pytest.mark.anyio
@requirement("mcpserver:resource:static")
async def test_read_static_resource(connect: Connect) -> None:
async def test_read_static_resource(connect: Connect, unstamped: Unstamp) -> None:
"""A function registered for a fixed URI is served at that URI with its return value as text."""
mcp = MCPServer("library")
@ -33,7 +34,7 @@ async def test_read_static_resource(connect: Connect) -> None:
async with connect(mcp) as client:
result = await client.read_resource("config://app")
assert result == snapshot(
assert unstamped(result) == snapshot(
ReadResourceResult(
contents=[TextResourceContents(uri="config://app", mime_type="text/plain", text="theme = dark")]
)
@ -41,7 +42,7 @@ async def test_read_static_resource(connect: Connect) -> None:
@requirement("mcpserver:resource:static")
async def test_list_static_and_templated_resources(connect: Connect) -> None:
async def test_list_static_and_templated_resources(connect: Connect, unstamped: Unstamp) -> None:
"""Statically-registered resources appear in resources/list; templated ones only in templates/list.
The name and description are derived from the function name and docstring; the MIME type
@ -63,7 +64,7 @@ async def test_list_static_and_templated_resources(connect: Connect) -> None:
resources = await client.list_resources()
templates = await client.list_resource_templates()
assert resources == snapshot(
assert unstamped(resources) == snapshot(
ListResourcesResult(
resources=[
Resource(
@ -75,7 +76,7 @@ async def test_list_static_and_templated_resources(connect: Connect) -> None:
]
)
)
assert templates == snapshot(
assert unstamped(templates) == snapshot(
ListResourceTemplatesResult(
resource_templates=[
ResourceTemplate(
@ -91,7 +92,7 @@ async def test_list_static_and_templated_resources(connect: Connect) -> None:
@requirement("mcpserver:resource:template")
@requirement("resources:read:template-vars")
async def test_read_templated_resource(connect: Connect) -> None:
async def test_read_templated_resource(connect: Connect, unstamped: Unstamp) -> None:
"""Reading a URI that matches a registered template invokes the function with the extracted parameters."""
mcp = MCPServer("library")
@ -103,7 +104,7 @@ async def test_read_templated_resource(connect: Connect) -> None:
async with connect(mcp) as client:
result = await client.read_resource("users://42/profile")
assert result == snapshot(
assert unstamped(result) == snapshot(
ReadResourceResult(
contents=[TextResourceContents(uri="users://42/profile", mime_type="text/plain", text="profile for 42")]
)
@ -152,7 +153,9 @@ async def test_resource_function_that_raises_is_surfaced_as_a_jsonrpc_error(conn
@requirement("mcpserver:resource:duplicate-name")
async def test_registering_a_duplicate_resource_uri_warns_and_keeps_the_first(connect: Connect) -> None:
async def test_registering_a_duplicate_resource_uri_warns_and_keeps_the_first(
connect: Connect, unstamped: Unstamp
) -> None:
"""Registering a second static resource at an already-used URI keeps the first registration.
The intended behaviour is rejection at registration time; MCPServer instead logs a warning
@ -178,6 +181,6 @@ async def test_registering_a_duplicate_resource_uri_warns_and_keeps_the_first(co
assert [resource.uri for resource in listed.resources] == ["config://app"]
assert listed.resources[0].name == "config_first"
assert result == snapshot(
assert unstamped(result) == snapshot(
ReadResourceResult(contents=[TextResourceContents(uri="config://app", mime_type="text/plain", text="first")])
)

47
tests/interaction/mcpserver/test_tools.py
View file

@ -20,6 +20,7 @@ from mcp import MCPError
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.shared.exceptions import UrlElicitationRequiredError
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._helpers import IncomingMessage
from tests.interaction._requirements import requirement
@ -28,7 +29,7 @@ pytestmark = pytest.mark.anyio
@requirement("tools:call:content:text")
async def test_call_tool_returns_text_content(connect: Connect) -> None:
async def test_call_tool_returns_text_content(connect: Connect, unstamped: Unstamp) -> None:
"""Arguments reach the tool function; its return value comes back as text content.
MCPServer also derives an output schema from the return annotation and attaches the
@ -43,11 +44,15 @@ async def test_call_tool_returns_text_content(connect: Connect) -> None:
async with connect(mcp) as client:
result = await client.call_tool("add", {"a": 2, "b": 3})
assert result == snapshot(CallToolResult(content=[TextContent(text="5")], structured_content={"result": "5"}))
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="5")], structured_content={"result": "5"})
)
@requirement("mcpserver:tool:schema-variants")
async def test_complex_parameter_types_are_validated_and_coerced_before_the_tool_runs(connect: Connect) -> None:
async def test_complex_parameter_types_are_validated_and_coerced_before_the_tool_runs(
connect: Connect, unstamped: Unstamp
) -> None:
"""Literal, nested-model, and constrained parameters are validated and coerced from the wire arguments.
The string "3" is coerced to `int` and the `point` dict to a `Point` instance before the function
@ -67,7 +72,7 @@ async def test_complex_parameter_types_are_validated_and_coerced_before_the_tool
async with connect(mcp) as client:
result = await client.call_tool("place", {"mode": "fast", "point": {"x": "3", "y": 4}, "count": 5})
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(
content=[TextContent(text="fast at (3, 4) x5")], structured_content={"result": "fast at (3, 4) x5"}
)
@ -76,7 +81,7 @@ async def test_complex_parameter_types_are_validated_and_coerced_before_the_tool
@requirement("mcpserver:tool:handler-throws")
@requirement("mcpserver:output-schema:skip-on-error")
async def test_call_tool_function_exception_becomes_error_result(connect: Connect) -> None:
async def test_call_tool_function_exception_becomes_error_result(connect: Connect, unstamped: Unstamp) -> None:
"""An exception raised by a tool function is returned as an is_error result, not a JSON-RPC error.
The function's `-> str` annotation gives the tool a derived output schema, but the error
@ -92,13 +97,13 @@ async def test_call_tool_function_exception_becomes_error_result(connect: Connec
async with connect(mcp) as client:
result = await client.call_tool("explode", {})
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="Error executing tool explode: boom")], is_error=True)
)
@requirement("mcpserver:tool:handler-throws")
async def test_call_tool_tool_error_becomes_error_result(connect: Connect) -> None:
async def test_call_tool_tool_error_becomes_error_result(connect: Connect, unstamped: Unstamp) -> None:
"""A ToolError raised by a tool function is returned as an is_error result, not a JSON-RPC error."""
mcp = MCPServer("errors")
@ -109,13 +114,13 @@ async def test_call_tool_tool_error_becomes_error_result(connect: Connect) -> No
async with connect(mcp) as client:
result = await client.call_tool("flux", {})
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="Error executing tool flux: flux capacitor offline")], is_error=True)
)
@requirement("mcpserver:tool:unknown-name")
async def test_call_tool_unknown_name_returns_error_result(connect: Connect) -> None:
async def test_call_tool_unknown_name_returns_error_result(connect: Connect, unstamped: Unstamp) -> None:
"""Calling a tool name that was never registered is reported as an is_error result.
The spec classifies unknown tools as a protocol error; see the divergence note on the
@ -130,12 +135,14 @@ async def test_call_tool_unknown_name_returns_error_result(connect: Connect) ->
async with connect(mcp) as client:
result = await client.call_tool("nope", {})
assert result == snapshot(CallToolResult(content=[TextContent(text="Unknown tool: nope")], is_error=True))
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="Unknown tool: nope")], is_error=True)
)
@requirement("mcpserver:tool:output-schema:model")
@requirement("tools:call:structured-content:text-mirror")
async def test_call_tool_model_return_becomes_structured_content(connect: Connect) -> None:
async def test_call_tool_model_return_becomes_structured_content(connect: Connect, unstamped: Unstamp) -> None:
"""A tool returning a pydantic model advertises the model's schema as the tool's output schema
and returns the model's fields as structured content alongside a serialised text block.
"""
@ -164,7 +171,7 @@ async def test_call_tool_model_return_becomes_structured_content(connect: Connec
"type": "object",
}
)
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(
content=[
TextContent(
@ -182,7 +189,7 @@ async def test_call_tool_model_return_becomes_structured_content(connect: Connec
@requirement("mcpserver:tool:output-schema:wrapped")
async def test_call_tool_list_return_is_wrapped_in_result_key(connect: Connect) -> None:
async def test_call_tool_list_return_is_wrapped_in_result_key(connect: Connect, unstamped: Unstamp) -> None:
"""A tool returning a list wraps the value under a "result" key in both the generated output
schema and the structured content.
"""
@ -204,7 +211,7 @@ async def test_call_tool_list_return_is_wrapped_in_result_key(connect: Connect)
"type": "object",
}
)
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(
content=[TextContent(text="2"), TextContent(text="3"), TextContent(text="5")],
structured_content={"result": [2, 3, 5]},
@ -279,7 +286,9 @@ async def test_tool_with_output_schema_returning_mismatched_structured_content_i
@requirement("mcpserver:tool:duplicate-name")
async def test_registering_a_duplicate_tool_name_warns_and_keeps_the_first(connect: Connect) -> None:
async def test_registering_a_duplicate_tool_name_warns_and_keeps_the_first(
connect: Connect, unstamped: Unstamp
) -> None:
"""Registering a second tool with an already-used name keeps the first registration.
The intended behaviour is rejection at registration time; MCPServer instead logs a warning
@ -304,14 +313,14 @@ async def test_registering_a_duplicate_tool_name_warns_and_keeps_the_first(conne
result = await client.call_tool("echo", {})
assert [tool.name for tool in listed.tools] == ["echo"]
assert result == snapshot(
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="first")], structured_content={"result": "first"})
)
@requirement("mcpserver:tool:naming-validation")
async def test_registering_a_tool_with_a_spec_invalid_name_warns_but_does_not_reject(
connect: Connect, caplog: pytest.LogCaptureFixture
connect: Connect, caplog: pytest.LogCaptureFixture, unstamped: Unstamp
) -> None:
"""A tool name that violates the SEP-986 rules logs a warning at registration but is still registered.
@ -340,7 +349,9 @@ async def test_registering_a_tool_with_a_spec_invalid_name_warns_but_does_not_re
result = await client.call_tool("bad name!", {})
assert [tool.name for tool in listed.tools] == ["bad name!"]
assert result == snapshot(CallToolResult(content=[TextContent(text="ok")], structured_content={"result": "ok"}))
assert unstamped(result) == snapshot(
CallToolResult(content=[TextContent(text="ok")], structured_content={"result": "ok"})
)
@requirement("mcpserver:tool:url-elicitation-error")

49
tests/interaction/transports/test_hosting_http_modern.py
View file

@ -22,6 +22,7 @@ from mcp_types import (
INVALID_PARAMS,
METHOD_NOT_FOUND,
MISSING_REQUIRED_CLIENT_CAPABILITY,
SERVER_INFO_META_KEY,
CallToolRequestParams,
CallToolResult,
DiscoverResult,
@ -77,7 +78,11 @@ def _meta_envelope() -> dict[str, object]:
def _server(*, on_meta: Callable[[dict[str, Any]], None] | None = None) -> Server:
"""A low-level server with one ``add`` tool for the raw-httpx2 tests below."""
"""A low-level server with one `add` tool for the raw-httpx2 tests below.
The explicit version gives the `_meta` serverInfo stamp every 2026 result
carries a non-empty value for the wire-level snapshots.
"""
async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
tool = Tool(name="add", input_schema={"type": "object"})
@ -91,7 +96,7 @@ def _server(*, on_meta: Callable[[dict[str, Any]], None] | None = None) -> Serve
on_meta(dict(ctx.meta))
return CallToolResult(content=[TextContent(text=str(params.arguments["a"] + params.arguments["b"]))])
return Server("modern", on_list_tools=list_tools, on_call_tool=call_tool)
return Server("modern", version="1.0.0", on_list_tools=list_tools, on_call_tool=call_tool)
@requirement("hosting:http:modern:tools-call-stateless")
@ -99,9 +104,10 @@ async def test_modern_tools_call_returns_result_type_complete_without_initialize
"""A 2026-07-28 tools/call is served without an initialize handshake and returns resultType: complete.
Spec-mandated under the draft transport: the per-request ``_meta`` envelope replaces initialize,
and ``resultType`` is the 2026 result-envelope discriminator (``complete`` for the monolith
result). Asserted at the wire because the SDK client never surfaces ``resultType`` and because
the absence of any prior request on the connection is the assertion.
`resultType` is the 2026 result-envelope discriminator (`complete` for the monolith
result), and the server identifies itself via the result `_meta` serverInfo stamp. Asserted at
the wire because the SDK client never surfaces `resultType` and because the absence of any
prior request on the connection is the assertion.
"""
body = {
"jsonrpc": "2.0",
@ -117,7 +123,12 @@ async def test_modern_tools_call_returns_result_type_complete_without_initialize
parsed = JSONRPCResponse.model_validate(response.json())
assert parsed.id == 1
assert parsed.result == snapshot(
{"content": [{"text": "5", "type": "text"}], "isError": False, "resultType": "complete"}
{
"content": [{"text": "5", "type": "text"}],
"isError": False,
"resultType": "complete",
"_meta": {"io.modelcontextprotocol/serverInfo": {"name": "modern", "version": "1.0.0"}},
}
)
@ -213,12 +224,13 @@ async def test_modern_handler_exception_maps_to_internal_error_without_leaking_t
@requirement("hosting:http:modern:discover-response-shape")
async def test_modern_server_discover_returns_capabilities_and_supported_versions() -> None:
"""A 2026-07-28 server/discover POST returns capabilities, serverInfo, and supportedVersions.
"""A 2026-07-28 server/discover POST returns capabilities and supportedVersions, with serverInfo in `_meta`.
Spec-mandated under the draft: server/discover is the 2026 advertisement method that replaces
the initialize-response payload, and ``supportedVersions`` is the field a client picks its
per-request envelope version from. Asserted at the wire because the SDK client never exposes
the raw result body.
per-request envelope version from. The server's identity is no longer a result-body field: it
travels as the io.modelcontextprotocol/serverInfo result `_meta` stamp. Asserted at the wire
because the SDK client never exposes the raw result body.
"""
body = {"jsonrpc": "2.0", "id": 1, "method": "server/discover", "params": {"_meta": _meta_envelope()}}
async with mounted_app(_server()) as (http, _):
@ -227,7 +239,8 @@ async def test_modern_server_discover_returns_capabilities_and_supported_version
assert response.status_code == 200
result = JSONRPCResponse.model_validate(response.json()).result
assert result["supportedVersions"] == snapshot(["2026-07-28"])
assert result["serverInfo"]["name"] == "modern"
assert "serverInfo" not in result
assert result["_meta"][SERVER_INFO_META_KEY] == {"name": "modern", "version": "1.0.0"}
assert "capabilities" in result
@ -282,7 +295,7 @@ async def test_modern_handler_raised_mcperror_maps_to_status_via_error_code_tabl
raise MCPError(
code=MISSING_REQUIRED_CLIENT_CAPABILITY,
message="sampling required",
data={"requiredCapabilities": ["sampling"]},
data={"requiredCapabilities": {"sampling": {}}},
)
server = _server()
@ -294,7 +307,7 @@ async def test_modern_handler_raised_mcperror_maps_to_status_via_error_code_tabl
assert response.status_code == 400
error = JSONRPCError.model_validate(response.json()).error
assert error.code == MISSING_REQUIRED_CLIENT_CAPABILITY
assert error.data == {"requiredCapabilities": ["sampling"]}
assert error.data == {"requiredCapabilities": {"sampling": {}}}
@requirement("hosting:http:modern:tools-call-stateless")
@ -340,7 +353,6 @@ async def test_pinned_client_stateless_tools_call_round_trips_against_the_modern
DiscoverResult(
supported_versions=[LATEST_MODERN_VERSION],
capabilities=ServerCapabilities(),
server_info=Implementation(name="srv", version="0"),
)
)
result = await session.call_tool(
@ -350,7 +362,12 @@ async def test_pinned_client_stateless_tools_call_round_trips_against_the_modern
)
assert result.model_dump(by_alias=True, mode="json", exclude_none=True) == snapshot(
{"content": [{"type": "text", "text": "5"}], "isError": False, "resultType": "complete"}
{
"_meta": {"io.modelcontextprotocol/serverInfo": {"name": "modern", "version": "1.0.0"}},
"content": [{"type": "text", "text": "5"}],
"isError": False,
"resultType": "complete",
}
)
# Exactly the tools/call POST and the implicit tools/list POST -- no initialize, no
@ -440,7 +457,6 @@ async def test_modern_client_mirrors_x_mcp_header_args_into_mcp_param_headers()
discover = DiscoverResult(
supported_versions=[LATEST_MODERN_VERSION],
capabilities=ServerCapabilities(),
server_info=Implementation(name="srv", version="0"),
)
with anyio.fail_after(5):
async with (
@ -487,7 +503,6 @@ async def test_modern_client_emits_no_param_headers_for_an_unlisted_tool() -> No
discover = DiscoverResult(
supported_versions=[LATEST_MODERN_VERSION],
capabilities=ServerCapabilities(),
server_info=Implementation(name="srv", version="0"),
)
with anyio.fail_after(5):
async with (
@ -541,7 +556,6 @@ async def test_modern_client_stops_mirroring_after_a_re_list_drops_the_tool() ->
discover = DiscoverResult(
supported_versions=[LATEST_MODERN_VERSION],
capabilities=ServerCapabilities(),
server_info=Implementation(name="srv", version="0"),
)
with anyio.fail_after(5):
async with (
@ -596,7 +610,6 @@ async def test_vendor_request_with_name_param_carries_mcp_name_on_the_wire() ->
discover = DiscoverResult(
supported_versions=[LATEST_MODERN_VERSION],
capabilities=ServerCapabilities(),
server_info=Implementation(name="srv", version="0"),
)
with anyio.fail_after(5):
async with (

1
tests/interaction/transports/test_stdio.py
View file

@ -90,6 +90,7 @@ async def test_tool_call_and_notification_round_trip_over_a_stdio_subprocess(
# Must exceed session time plus the patched PROCESS_TERMINATION_TIMEOUT (20s).
with anyio.fail_after(30):
async with Client(transport, mode="legacy", logging_callback=collect) as client:
assert client.server_info is not None
assert client.server_info.name == "stdio-echo"
result = await client.call_tool("echo", {"text": "across\nprocesses"})

Some files were not shown because too many files have changed in this diff Show more