mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 07:09:11 +02:00
Update tests for the removed sampling and roots server API
This commit is contained in:
Jeremiah Lowin
parent
e056a3946e
commit
704b74b3ab
9 changed files with 141 additions and 507 deletions
|
|
@ -21,7 +21,6 @@ from fastmcp.client import Client
|
|||
from fastmcp.client.mixins.tools import _parse_call_tool_result
|
||||
from fastmcp.client.transports import FastMCPTransport
|
||||
from fastmcp.exceptions import PromptError, ResourceError, ToolError
|
||||
from fastmcp.server.sampling.run import SamplingTool, execute_tools
|
||||
from fastmcp.server.server import FastMCP
|
||||
|
||||
|
||||
|
|
@ -416,37 +415,3 @@ class TestLogLevel:
|
|||
for record in caplog.records
|
||||
)
|
||||
|
||||
async def test_sampling_tool_error_with_custom_log_level(self, caplog):
|
||||
"""ToolError with custom log_level in sampling should log at specified level."""
|
||||
|
||||
async def custom_level_sampling_tool(x: int) -> int:
|
||||
raise ToolError("Expected sampling error", log_level=logging.WARNING)
|
||||
|
||||
tool = SamplingTool.from_function(custom_level_sampling_tool)
|
||||
tool_use = ToolUseContent(
|
||||
type="tool_use",
|
||||
id="test-id",
|
||||
name="custom_level_sampling_tool",
|
||||
input={"x": 42},
|
||||
)
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
results = await execute_tools(
|
||||
tool_calls=[tool_use],
|
||||
tool_map={"custom_level_sampling_tool": tool},
|
||||
mask_error_details=False,
|
||||
)
|
||||
|
||||
assert len(results) == 1
|
||||
assert results[0].is_error
|
||||
assert "Expected sampling error" in results[0].content[0].text # type: ignore
|
||||
assert any(
|
||||
"Error calling sampling tool" in record.message
|
||||
and record.levelname == "WARNING"
|
||||
for record in caplog.records
|
||||
)
|
||||
assert not any(
|
||||
"Error calling sampling tool" in record.message
|
||||
and record.levelname == "ERROR"
|
||||
for record in caplog.records
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,16 +1,23 @@
|
|||
import pytest
|
||||
from mcp_types import Root
|
||||
|
||||
from fastmcp import Client, Context, FastMCP
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fastmcp_server():
|
||||
"""A server that issues a handshake-era `roots/list` request.
|
||||
|
||||
`Context` has no `list_roots()` — server-initiated requests are not part of
|
||||
FastMCP's server API. This server reaches the SDK session directly to stand
|
||||
in for a legacy upstream, so the client's `roots=` handling stays covered.
|
||||
"""
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
async def list_roots(context: Context) -> list[str]:
|
||||
roots = await context.list_roots()
|
||||
return [str(r.uri) for r in roots]
|
||||
result = await context.session.list_roots()
|
||||
return [str(r.uri) for r in result.roots]
|
||||
|
||||
return mcp
|
||||
|
||||
|
|
@ -36,10 +43,27 @@ class TestClientRoots:
|
|||
|
||||
@pytest.mark.parametrize("roots", [["file://x/y/z", "file://x/y/z"]])
|
||||
async def test_valid_roots(self, fastmcp_server: FastMCP, roots: list[str]):
|
||||
# ctx.list_roots is a legacy-era server-initiated feature.
|
||||
# `roots/list` is a server-initiated request, so it only exists on the
|
||||
# handshake era; SEP-2577 removed it from the modern protocol.
|
||||
async with Client(fastmcp_server, mode="legacy", roots=roots) as client:
|
||||
result = await client.call_tool("list_roots", {})
|
||||
assert result.data == [
|
||||
"file://x/y/z",
|
||||
"file://x/y/z",
|
||||
]
|
||||
|
||||
async def test_roots_handler_answers_a_legacy_server(self, fastmcp_server: FastMCP):
|
||||
"""A callable `roots=` handler still answers a legacy server's request."""
|
||||
calls: list[object] = []
|
||||
|
||||
async def roots_handler(ctx) -> list[Root]:
|
||||
calls.append(ctx)
|
||||
return [Root(uri="file://from/handler")] # ty: ignore[invalid-argument-type]
|
||||
|
||||
async with Client(
|
||||
fastmcp_server, mode="legacy", roots=roots_handler
|
||||
) as client:
|
||||
result = await client.call_tool("list_roots", {})
|
||||
|
||||
assert len(calls) == 1
|
||||
assert result.data == ["file://from/handler"]
|
||||
|
|
|
|||
|
|
@ -1,71 +1,95 @@
|
|||
import json
|
||||
from typing import cast
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import mcp_types
|
||||
import pytest
|
||||
from mcp_types import TextContent
|
||||
from pydantic_core import to_json
|
||||
|
||||
from fastmcp import Client, Context, FastMCP
|
||||
from fastmcp.client.sampling import RequestContext, SamplingMessage, SamplingParams
|
||||
from fastmcp.server.sampling import SamplingResult, SamplingTool
|
||||
from fastmcp.utilities.types import Image
|
||||
|
||||
|
||||
async def _sample(
|
||||
context: Context,
|
||||
messages: list[SamplingMessage],
|
||||
*,
|
||||
system_prompt: str | None = None,
|
||||
) -> str:
|
||||
"""Issue a handshake-era `sampling/createMessage` request from a server.
|
||||
|
||||
`Context` has no `sample()` — server-initiated sampling is not part of
|
||||
FastMCP's server API. These tests cover the *client* side, which must keep
|
||||
answering a legacy server, so the stand-in server reaches the SDK session
|
||||
directly.
|
||||
"""
|
||||
result = await context.session.create_message(
|
||||
messages=messages,
|
||||
system_prompt=system_prompt,
|
||||
max_tokens=512,
|
||||
related_request_id=context.origin_request_id,
|
||||
)
|
||||
assert isinstance(result.content, TextContent)
|
||||
return result.content.text
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fastmcp_server():
|
||||
mcp = FastMCP()
|
||||
|
||||
@mcp.tool
|
||||
async def simple_sample(message: str, context: Context) -> str:
|
||||
result = await context.sample("Hello, world!")
|
||||
assert isinstance(result, SamplingResult)
|
||||
assert result.text is not None
|
||||
return result.text
|
||||
return await _sample(
|
||||
context,
|
||||
[
|
||||
SamplingMessage(
|
||||
role="user",
|
||||
content=TextContent(type="text", text="Hello, world!"),
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
@mcp.tool
|
||||
async def sample_with_system_prompt(message: str, context: Context) -> str:
|
||||
result = await context.sample("Hello, world!", system_prompt="You love FastMCP")
|
||||
assert isinstance(result, SamplingResult)
|
||||
assert result.text is not None
|
||||
return result.text
|
||||
return await _sample(
|
||||
context,
|
||||
[
|
||||
SamplingMessage(
|
||||
role="user",
|
||||
content=TextContent(type="text", text="Hello, world!"),
|
||||
)
|
||||
],
|
||||
system_prompt="You love FastMCP",
|
||||
)
|
||||
|
||||
@mcp.tool
|
||||
async def sample_with_messages(message: str, context: Context) -> str:
|
||||
result = await context.sample(
|
||||
return await _sample(
|
||||
context,
|
||||
[
|
||||
"Hello!",
|
||||
SamplingMessage(
|
||||
content=TextContent(
|
||||
type="text", text="How can I assist you today?"
|
||||
),
|
||||
role="assistant",
|
||||
role="user", content=TextContent(type="text", text="Hello!")
|
||||
),
|
||||
]
|
||||
SamplingMessage(
|
||||
role="assistant",
|
||||
content=TextContent(type="text", text="How can I assist you today?"),
|
||||
),
|
||||
],
|
||||
)
|
||||
assert isinstance(result, SamplingResult)
|
||||
assert result.text is not None
|
||||
return result.text
|
||||
|
||||
@mcp.tool
|
||||
async def sample_with_image(image_bytes: bytes, context: Context) -> str:
|
||||
image = Image(data=image_bytes)
|
||||
|
||||
result = await context.sample(
|
||||
return await _sample(
|
||||
context,
|
||||
[
|
||||
SamplingMessage(
|
||||
content=TextContent(type="text", text="What's in this image?"),
|
||||
role="user",
|
||||
),
|
||||
SamplingMessage(
|
||||
content=image.to_image_content(),
|
||||
role="user",
|
||||
),
|
||||
]
|
||||
SamplingMessage(content=image.to_image_content(), role="user"),
|
||||
],
|
||||
)
|
||||
assert isinstance(result, SamplingResult)
|
||||
assert result.text is not None
|
||||
return result.text
|
||||
|
||||
return mcp
|
||||
|
||||
|
|
@ -123,26 +147,6 @@ async def test_sampling_with_messages(fastmcp_server: FastMCP):
|
|||
assert result.data == "I need to think."
|
||||
|
||||
|
||||
async def test_sampling_with_fallback(fastmcp_server: FastMCP):
|
||||
openai_sampling_handler = AsyncMock(return_value="But I need to think")
|
||||
|
||||
fastmcp_server = FastMCP(
|
||||
sampling_handler=openai_sampling_handler,
|
||||
)
|
||||
|
||||
@fastmcp_server.tool
|
||||
async def sample_with_fallback(context: Context) -> str:
|
||||
sampling_result = await context.sample("Do not think.")
|
||||
return cast(TextContent, sampling_result).text
|
||||
|
||||
client = Client(fastmcp_server)
|
||||
|
||||
async with client:
|
||||
call_tool_result = await client.call_tool("sample_with_fallback")
|
||||
|
||||
assert call_tool_result.data == "But I need to think"
|
||||
|
||||
|
||||
async def test_sampling_with_image(fastmcp_server: FastMCP):
|
||||
def sampling_handler(
|
||||
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
|
||||
|
|
@ -193,8 +197,6 @@ class TestSamplingDefaultCapabilities:
|
|||
{"sampling": {"tools": {}}}, ensuring compatibility with servers
|
||||
that don't recognize the tools sub-field (e.g. older Java MCP SDK).
|
||||
"""
|
||||
import mcp_types
|
||||
|
||||
server = FastMCP()
|
||||
|
||||
def handler(
|
||||
|
|
@ -209,8 +211,6 @@ class TestSamplingDefaultCapabilities:
|
|||
|
||||
async def test_set_sampling_callback_default_capabilities_omit_tools(self):
|
||||
"""set_sampling_callback should also default to no tools capability."""
|
||||
import mcp_types
|
||||
|
||||
server = FastMCP()
|
||||
client = Client(server)
|
||||
client.set_sampling_callback(lambda msgs, params, ctx: "ok")
|
||||
|
|
@ -220,8 +220,6 @@ class TestSamplingDefaultCapabilities:
|
|||
|
||||
async def test_explicit_tools_capability_is_preserved(self):
|
||||
"""Explicitly passing tools capability should be respected."""
|
||||
import mcp_types
|
||||
|
||||
server = FastMCP()
|
||||
|
||||
def handler(
|
||||
|
|
@ -238,118 +236,3 @@ class TestSamplingDefaultCapabilities:
|
|||
caps = client._session_kwargs["sampling_capabilities"]
|
||||
assert isinstance(caps, mcp_types.SamplingCapability)
|
||||
assert caps.tools is not None
|
||||
|
||||
|
||||
class TestSamplingWithTools:
|
||||
"""Tests for sampling with tools functionality."""
|
||||
|
||||
async def test_sampling_with_tools_requires_capability(self):
|
||||
"""Test that sampling with tools raises error when client lacks capability."""
|
||||
import mcp_types
|
||||
|
||||
from fastmcp.exceptions import ToolError
|
||||
|
||||
server = FastMCP()
|
||||
|
||||
def search(query: str) -> str:
|
||||
"""Search the web."""
|
||||
return f"Results for: {query}"
|
||||
|
||||
@server.tool
|
||||
async def sample_with_tool(context: Context) -> str:
|
||||
# This should fail because the client doesn't advertise tools capability
|
||||
result = await context.sample(
|
||||
messages="Search for Python tutorials",
|
||||
tools=[search],
|
||||
)
|
||||
return str(result)
|
||||
|
||||
def sampling_handler(
|
||||
messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext
|
||||
) -> str:
|
||||
return "Response"
|
||||
|
||||
# Explicitly disable tools capability by passing SamplingCapability without tools
|
||||
async with Client(
|
||||
server,
|
||||
mode="legacy",
|
||||
sampling_handler=sampling_handler,
|
||||
sampling_capabilities=mcp_types.SamplingCapability(), # No tools
|
||||
) as client:
|
||||
with pytest.raises(ToolError, match="sampling.tools capability"):
|
||||
await client.call_tool("sample_with_tool", {})
|
||||
|
||||
async def test_sampling_with_tools_fallback_handler_can_return_string(self):
|
||||
"""Test that fallback handler can return a string even when tools are provided.
|
||||
|
||||
The LLM might choose not to use any tools and just return a text response.
|
||||
"""
|
||||
# This handler returns a string - valid even when tools are provided
|
||||
simple_handler = AsyncMock(return_value="Direct response without tools")
|
||||
|
||||
mcp = FastMCP(sampling_handler=simple_handler)
|
||||
|
||||
def search(query: str) -> str:
|
||||
"""Search the web."""
|
||||
return f"Results for: {query}"
|
||||
|
||||
@mcp.tool
|
||||
async def sample_with_tool(context: Context) -> str:
|
||||
result = await context.sample(
|
||||
messages="Search for Python tutorials",
|
||||
tools=[search],
|
||||
)
|
||||
return result.text or "no text"
|
||||
|
||||
# Client without sampling handler - will use server's fallback
|
||||
async with Client(mcp) as client:
|
||||
result = await client.call_tool("sample_with_tool", {})
|
||||
|
||||
# Handler returned string directly, which is treated as final text response
|
||||
assert result.data == "Direct response without tools"
|
||||
|
||||
def test_sampling_tool_schema(self):
|
||||
"""Test that SamplingTool generates correct schema."""
|
||||
|
||||
def search(query: str, limit: int = 10) -> str:
|
||||
"""Search the web for results."""
|
||||
return f"Results for: {query}"
|
||||
|
||||
tool = SamplingTool.from_function(search)
|
||||
assert tool.name == "search"
|
||||
assert tool.description == "Search the web for results."
|
||||
assert "query" in tool.parameters.get("properties", {})
|
||||
assert "limit" in tool.parameters.get("properties", {})
|
||||
|
||||
async def test_sampling_tool_run(self):
|
||||
"""Test that SamplingTool.run() executes correctly."""
|
||||
|
||||
def add(a: int, b: int) -> int:
|
||||
"""Add two numbers."""
|
||||
return a + b
|
||||
|
||||
tool = SamplingTool.from_function(add)
|
||||
result = await tool.run({"a": 5, "b": 3})
|
||||
assert result == 8
|
||||
|
||||
async def test_sampling_tool_run_async(self):
|
||||
"""Test that SamplingTool.run() works with async functions."""
|
||||
|
||||
async def async_multiply(a: int, b: int) -> int:
|
||||
"""Multiply two numbers."""
|
||||
return a * b
|
||||
|
||||
tool = SamplingTool.from_function(async_multiply)
|
||||
result = await tool.run({"a": 4, "b": 7})
|
||||
assert result == 28
|
||||
|
||||
def test_tool_choice_parameter(self):
|
||||
"""Test that tool_choice parameter accepts string literals."""
|
||||
from fastmcp.server.context import ToolChoiceOption
|
||||
|
||||
# Verify ToolChoiceOption type accepts the valid string values
|
||||
choices: list[ToolChoiceOption] = ["auto", "required", "none"]
|
||||
assert len(choices) == 3
|
||||
assert "auto" in choices
|
||||
assert "required" in choices
|
||||
assert "none" in choices
|
||||
|
|
|
|||
|
|
@ -134,16 +134,6 @@ async def test_tool_with_progress(ctx: Context) -> str:
|
|||
return "Progress test complete."
|
||||
|
||||
|
||||
@server.tool(name="test_sampling")
|
||||
async def test_sampling(prompt: str, ctx: Context) -> str:
|
||||
"""Requests LLM sampling via the client."""
|
||||
result = await ctx.sample(
|
||||
messages=[prompt],
|
||||
result_type=str,
|
||||
)
|
||||
return f"Sampling result: {result}"
|
||||
|
||||
|
||||
class _UserInfo(BaseModel):
|
||||
username: str
|
||||
email: str
|
||||
|
|
|
|||
|
|
@ -147,10 +147,6 @@ def mcp_server(recording_middleware):
|
|||
async def log_tool(context: Context) -> None:
|
||||
await context.info(message="test log")
|
||||
|
||||
@mcp.tool
|
||||
async def sample_tool(context: Context) -> None:
|
||||
await context.sample("hello")
|
||||
|
||||
mcp.add_middleware(recording_middleware)
|
||||
|
||||
# Register a progress notification handler (v2 API: (ctx, params)).
|
||||
|
|
|
|||
|
|
@ -149,10 +149,6 @@ def mcp_server(recording_middleware):
|
|||
async def log_tool(context: Context) -> None:
|
||||
await context.info(message="test log")
|
||||
|
||||
@mcp.tool
|
||||
async def sample_tool(context: Context) -> None:
|
||||
await context.sample("hello")
|
||||
|
||||
mcp.add_middleware(recording_middleware)
|
||||
|
||||
# Register a progress notification handler (v2 API: (ctx, params)).
|
||||
|
|
@ -205,10 +201,6 @@ class TestNestedMiddlewareHooks:
|
|||
async def log_tool(context: Context) -> None:
|
||||
await context.info(message="test log")
|
||||
|
||||
@mcp.tool
|
||||
async def sample_tool(context: Context) -> None:
|
||||
await context.sample("hello")
|
||||
|
||||
mcp.add_middleware(nested_middleware)
|
||||
|
||||
return mcp
|
||||
|
|
@ -510,7 +502,7 @@ class TestProxyServer:
|
|||
async with Client(proxy_server) as client:
|
||||
await client.list_tools()
|
||||
|
||||
assert TAGS == [{"add-tool"}, set(), set(), set()]
|
||||
assert TAGS == [{"add-tool"}, set(), set()]
|
||||
|
||||
|
||||
class TestToolCallDenial:
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ from mcp_types import (
|
|||
LoggingLevel,
|
||||
ModelHint,
|
||||
ModelPreferences,
|
||||
Root,
|
||||
TextContent,
|
||||
)
|
||||
from pydantic import BaseModel, Field
|
||||
|
|
@ -58,6 +59,34 @@ class TestProxyClientEraDefault:
|
|||
assert cast(Client, factory()).mode == "auto"
|
||||
|
||||
|
||||
async def _backend_list_roots(context: Context) -> list[Root]:
|
||||
"""Issue a handshake-era `roots/list` from a backend server.
|
||||
|
||||
`Context` has no `list_roots()`: server-initiated requests are not part of
|
||||
FastMCP's server API. These helpers reach the SDK session directly to stand
|
||||
in for a legacy upstream, which is the only thing the proxy relay forwards.
|
||||
"""
|
||||
result = await context.session.list_roots()
|
||||
return result.roots
|
||||
|
||||
|
||||
async def _backend_sample(context: Context) -> str:
|
||||
"""Issue a handshake-era `sampling/createMessage` from a backend server."""
|
||||
result = await context.session.create_message(
|
||||
messages=[
|
||||
SamplingMessage(
|
||||
role="user", content=TextContent(type="text", text="Hello, world!")
|
||||
)
|
||||
],
|
||||
system_prompt="You love FastMCP",
|
||||
temperature=0.5,
|
||||
max_tokens=100,
|
||||
model_preferences=ModelPreferences(hints=[ModelHint(name="gpt-4o")]),
|
||||
related_request_id=context.origin_request_id,
|
||||
)
|
||||
return result.content.text if isinstance(result.content, TextContent) else ""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fastmcp_server():
|
||||
mcp = FastMCP("TestServer")
|
||||
|
|
@ -68,21 +97,13 @@ def fastmcp_server():
|
|||
|
||||
@mcp.tool
|
||||
async def list_roots(context: Context) -> list[str]:
|
||||
roots = await context.list_roots()
|
||||
return [str(r.uri) for r in roots]
|
||||
return [str(r.uri) for r in await _backend_list_roots(context)]
|
||||
|
||||
@mcp.tool
|
||||
async def sampling(
|
||||
context: Context,
|
||||
) -> str:
|
||||
result = await context.sample(
|
||||
"Hello, world!",
|
||||
system_prompt="You love FastMCP",
|
||||
temperature=0.5,
|
||||
max_tokens=100,
|
||||
model_preferences="gpt-4o",
|
||||
)
|
||||
return result.text or ""
|
||||
return await _backend_sample(context)
|
||||
|
||||
@dataclass
|
||||
class Person:
|
||||
|
|
@ -514,17 +535,16 @@ def roots_backend_server():
|
|||
|
||||
@mcp.resource("data://roots")
|
||||
async def roots_resource(context: Context) -> list[str]:
|
||||
roots = await context.list_roots()
|
||||
return [str(r.uri) for r in roots]
|
||||
return [str(r.uri) for r in await _backend_list_roots(context)]
|
||||
|
||||
@mcp.resource("data://roots/{key}")
|
||||
async def roots_template(key: str, context: Context) -> str:
|
||||
roots = await context.list_roots()
|
||||
roots = await _backend_list_roots(context)
|
||||
return ", ".join(f"{key}:{r.uri}" for r in roots)
|
||||
|
||||
@mcp.prompt
|
||||
async def roots_prompt(context: Context) -> str:
|
||||
roots = await context.list_roots()
|
||||
roots = await _backend_list_roots(context)
|
||||
return ", ".join(str(r.uri) for r in roots)
|
||||
|
||||
return mcp
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ from fastmcp.server.context import (
|
|||
reset_transport,
|
||||
set_transport,
|
||||
)
|
||||
from fastmcp.server.sampling.run import _parse_model_preferences
|
||||
from fastmcp.server.server import FastMCP
|
||||
|
||||
|
||||
|
|
@ -17,28 +16,6 @@ def context():
|
|||
return Context(fastmcp=FastMCP())
|
||||
|
||||
|
||||
class TestParseModelPreferences:
|
||||
def test_parse_model_preferences_string(self, context):
|
||||
mp = _parse_model_preferences("claude-haiku-4-5")
|
||||
assert isinstance(mp, ModelPreferences)
|
||||
assert mp.hints is not None
|
||||
assert mp.hints[0].name == "claude-haiku-4-5"
|
||||
|
||||
def test_parse_model_preferences_list(self, context):
|
||||
mp = _parse_model_preferences(["claude-haiku-4-5", "claude"])
|
||||
assert isinstance(mp, ModelPreferences)
|
||||
assert mp.hints is not None
|
||||
assert [h.name for h in mp.hints] == ["claude-haiku-4-5", "claude"]
|
||||
|
||||
def test_parse_model_preferences_object(self, context):
|
||||
obj = ModelPreferences(hints=[])
|
||||
assert _parse_model_preferences(obj) is obj
|
||||
|
||||
def test_parse_model_preferences_invalid_type(self, context):
|
||||
with pytest.raises(ValueError):
|
||||
_parse_model_preferences(model_preferences=123) # pyright: ignore[reportArgumentType] # type: ignore[invalid-argument-type] # ty:ignore[invalid-argument-type]
|
||||
|
||||
|
||||
class TestSessionId:
|
||||
def test_session_id_with_http_headers(self, context):
|
||||
"""Test that session_id returns the value from mcp-session-id header."""
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ from mcp.shared.exceptions import MCPError
|
|||
from pydantic import FileUrl
|
||||
|
||||
from fastmcp import Client as FastMCPClient
|
||||
from fastmcp import Context, FastMCP, settings
|
||||
from fastmcp import Context, FastMCP
|
||||
from fastmcp.exceptions import PromptError, ResourceError
|
||||
from fastmcp.server.elicitation import AcceptedElicitation
|
||||
from fastmcp.server.middleware import Middleware
|
||||
|
|
@ -215,16 +215,6 @@ def push_server() -> FastMCP:
|
|||
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")
|
||||
|
|
@ -264,31 +254,12 @@ async def test_elicit_works_on_legacy(push_server):
|
|||
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 def test_elicit_degrades_on_modern(push_server, mode):
|
||||
"""Elicitation is a server-initiated request, removed at 2026-07-28
|
||||
(SEP-2577), so a tool that uses it must degrade to a surfaced error rather
|
||||
than hang or crash the connection. The connection survives: a subsequent
|
||||
normal call still works.
|
||||
"""
|
||||
async with SDKClient(
|
||||
_server(push_server),
|
||||
|
|
@ -297,230 +268,46 @@ async def test_push_features_degrade_on_modern(push_server, mode, tool):
|
|||
sampling_callback=_sampling_cb,
|
||||
list_roots_callback=_roots_cb,
|
||||
) as client:
|
||||
result = await client.call_tool(tool, {})
|
||||
result = await client.call_tool("do_elicit", {})
|
||||
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 def test_elicit_degradation_message_is_clear_on_modern(push_server):
|
||||
"""FastMCP era-gates elicit: on a 2026-07-28 connection it raises a clear,
|
||||
era-aware error before hitting the wire, instead of the SDK's opaque
|
||||
'Method not found' (sdk-feedback.md #10).
|
||||
"""
|
||||
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, {})
|
||||
result = await client.call_tool("do_elicit", {})
|
||||
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
|
||||
# 3a-bis. Sampling and roots are not in the server API at all
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@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("name", ["sample", "sample_step", "list_roots"])
|
||||
def test_removed_server_initiated_methods_are_absent(name):
|
||||
"""FastMCP 4 targets the modern protocol, so the capabilities SEP-2577
|
||||
removed are not in the server-authoring API — not deprecated, not era-gated,
|
||||
absent. A server that calls them fails at attribute lookup, in every era.
|
||||
"""
|
||||
assert not hasattr(Context, name)
|
||||
|
||||
|
||||
@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("kwarg", ["sampling_handler", "sampling_handler_behavior"])
|
||||
def test_server_sampling_handler_kwargs_are_rejected(kwarg):
|
||||
"""The server-side sampling handler existed only to answer `ctx.sample()`."""
|
||||
with pytest.raises(TypeError, match="SEP-2577"):
|
||||
FastMCP("gone", **{kwarg: None})
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", MODERN_MODES)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue