mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-11 08:09:10 +02:00
db6d7a8a61* fix: elicitation scalar return, resource auto-serialization, Client.new() state, prompt errors
- Auto-wrap scalar elicitation responses for ScalarElicitationType schemas
so handlers can return T directly for ctx.elicit("msg", str/int/float)
- Auto-serialize dict/int/float/bool/None resource returns to JSON text
instead of crashing with TypeError
- Reset _task_registry and _submitted_task_ids in Client.new() so cloned
clients have independent task tracking state
- Include original error message in prompt render errors (matching tool
error behavior)
Fixes #3856
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* Fix misleading comment and add list/tuple auto-serialization for resources
The comment said "list/tuple of primitives" but the isinstance check
didn't include list or tuple. Now it does, and the comment matches.
🤖 Generated with Claude Code
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* Fix TaskNotificationHandler binding in Client.new() and meta forwarding for JSON resources
Two review-identified bugs:
1. Client.new() shallow-copies _session_kwargs, so the cloned client's
TaskNotificationHandler still dispatches to the original client.
Fix: create a fresh _session_kwargs dict with a new handler bound
to the new client.
2. convert_result() for dict/int/float/bool/None fell through to
ResourceResult(raw_value) which lost component meta (CSP, permissions).
The str/bytes path correctly wrapped in ResourceContent with meta.
Fix: explicitly serialize JSON-native types and wrap with meta,
matching the str/bytes path. Other types still fall through for
error handling.
🤖 Generated with Claude Code
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* Preserve custom message handlers in Client.new()
Only replace the message handler with a new TaskNotificationHandler
if the current handler IS a TaskNotificationHandler. If the user
provided a custom message_handler, preserve it in the clone.
🤖 Generated with Claude Code
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* fix client.new and add regression tests
* Honor declared MIME type for auto-serialized JSON resources
* Fix static analysis: remove unused StdioTransport import, fix ty:ignore comment
* Exclude list[ResourceContent] from JSON auto-serialization path
A bare list[ResourceContent] would match the isinstance(list) check
and get JSON-serialized instead of passing through to ResourceResult
normalization. Check for ResourceContent items first.
🤖 Generated with Claude Code
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
335 lines
11 KiB
Python
335 lines
11 KiB
Python
import pytest
|
|
from pydantic import AnyUrl, BaseModel
|
|
|
|
from fastmcp.resources.base import ResourceContent
|
|
from fastmcp.resources.function_resource import FunctionResource
|
|
|
|
|
|
class TestFunctionResource:
|
|
"""Test FunctionResource functionality."""
|
|
|
|
def test_function_resource_creation(self):
|
|
"""Test creating a FunctionResource."""
|
|
|
|
def my_func() -> str:
|
|
return "test content"
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("fn://test"),
|
|
name="test",
|
|
description="test function",
|
|
fn=my_func,
|
|
)
|
|
assert str(resource.uri) == "fn://test"
|
|
assert resource.name == "test"
|
|
assert resource.description == "test function"
|
|
assert resource.mime_type == "text/plain" # default
|
|
assert resource.fn == my_func
|
|
|
|
async def test_read_text(self):
|
|
"""Test reading text from a FunctionResource."""
|
|
|
|
def get_data() -> str:
|
|
return "Hello, world!"
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=get_data,
|
|
)
|
|
# read() returns raw value
|
|
result = await resource.read()
|
|
assert result == "Hello, world!"
|
|
|
|
# _read() converts to ResourceResult
|
|
result = await resource._read()
|
|
assert len(result.contents) == 1
|
|
assert result.contents[0].content == "Hello, world!"
|
|
assert result.contents[0].mime_type == "text/plain"
|
|
|
|
async def test_read_binary(self):
|
|
"""Test reading binary data from a FunctionResource."""
|
|
|
|
def get_data() -> bytes:
|
|
return b"Hello, world!"
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=get_data,
|
|
)
|
|
# read() returns raw value
|
|
result = await resource.read()
|
|
assert result == b"Hello, world!"
|
|
|
|
# _read() converts to ResourceResult
|
|
result = await resource._read()
|
|
assert result.contents[0].content == b"Hello, world!"
|
|
|
|
async def test_dict_return_auto_serializes(self):
|
|
"""Returning dict from read() auto-serializes to JSON."""
|
|
|
|
def get_data() -> dict:
|
|
return {"key": "value"}
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=get_data,
|
|
)
|
|
# read() returns raw value (no type checking at runtime)
|
|
result = await resource.read()
|
|
assert result == {"key": "value"}
|
|
|
|
# _read() auto-serializes dict to JSON text
|
|
resource_result = await resource._read()
|
|
assert len(resource_result.contents) == 1
|
|
assert '"key"' in str(resource_result.contents[0].content)
|
|
|
|
async def test_error_handling(self):
|
|
"""Test error handling in FunctionResource."""
|
|
|
|
def failing_func() -> str:
|
|
raise ValueError("Test error")
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=failing_func,
|
|
)
|
|
with pytest.raises(ValueError, match="Test error"):
|
|
await resource.read()
|
|
|
|
async def test_basemodel_return_raises_type_error(self):
|
|
"""Returning BaseModel from read() raises TypeError - use ResourceResult."""
|
|
|
|
class MyModel(BaseModel):
|
|
name: str
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=lambda: MyModel(name="test"),
|
|
)
|
|
# read() returns raw value (no type checking at runtime)
|
|
raw_result = await resource.read()
|
|
assert isinstance(raw_result, MyModel)
|
|
|
|
# _read() raises TypeError - must return str, bytes, or ResourceResult
|
|
with pytest.raises(TypeError, match="must be str, bytes, or list"):
|
|
await resource._read()
|
|
|
|
async def test_custom_type_return_raises_type_error(self):
|
|
"""Returning custom type from read() raises TypeError - use ResourceResult."""
|
|
|
|
class CustomData:
|
|
def __str__(self) -> str:
|
|
return "custom data"
|
|
|
|
def get_data() -> CustomData:
|
|
return CustomData()
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=get_data,
|
|
)
|
|
# read() returns raw value (no type checking at runtime)
|
|
raw_result = await resource.read()
|
|
assert isinstance(raw_result, CustomData)
|
|
|
|
# _read() raises TypeError - must return str, bytes, or ResourceResult
|
|
with pytest.raises(TypeError, match="must be str, bytes, or list"):
|
|
await resource._read()
|
|
|
|
async def test_async_read_text(self):
|
|
"""Test reading text from async FunctionResource."""
|
|
|
|
async def get_data() -> str:
|
|
return "Hello, world!"
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=get_data,
|
|
)
|
|
# read() returns raw value
|
|
result = await resource.read()
|
|
assert result == "Hello, world!"
|
|
|
|
# _read() converts to ResourceResult
|
|
result = await resource._read()
|
|
assert result.contents[0].content == "Hello, world!"
|
|
assert result.contents[0].mime_type == "text/plain"
|
|
|
|
async def test_resource_content_text(self):
|
|
"""Test returning ResourceContent with text content."""
|
|
|
|
def get_data() -> ResourceContent:
|
|
return ResourceContent(
|
|
content="Hello, world!",
|
|
mime_type="text/html",
|
|
meta={"csp": "script-src 'self'"},
|
|
)
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=get_data,
|
|
)
|
|
result = await resource.read()
|
|
assert isinstance(result, ResourceContent)
|
|
assert result.content == "Hello, world!"
|
|
assert result.mime_type == "text/html"
|
|
assert result.meta == {"csp": "script-src 'self'"}
|
|
|
|
async def test_resource_content_binary(self):
|
|
"""Test returning ResourceContent with binary content."""
|
|
|
|
def get_data() -> ResourceContent:
|
|
return ResourceContent(
|
|
content=b"\x00\x01\x02",
|
|
mime_type="application/octet-stream",
|
|
)
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=get_data,
|
|
)
|
|
result = await resource.read()
|
|
assert isinstance(result, ResourceContent)
|
|
assert result.content == b"\x00\x01\x02"
|
|
assert result.mime_type == "application/octet-stream"
|
|
assert result.meta is None
|
|
|
|
async def test_resource_content_without_meta(self):
|
|
"""Test that ResourceContent auto-sets mime_type defaults."""
|
|
content = ResourceContent(content="plain text")
|
|
assert content.content == "plain text"
|
|
assert content.mime_type == "text/plain" # Auto-set for string content
|
|
assert content.meta is None
|
|
|
|
async def test_async_resource_content(self):
|
|
"""Test async function returning ResourceContent."""
|
|
|
|
async def get_data() -> ResourceContent:
|
|
return ResourceContent(
|
|
content="async content",
|
|
meta={"key": "value"},
|
|
)
|
|
|
|
resource = FunctionResource(
|
|
uri=AnyUrl("function://test"),
|
|
name="test",
|
|
fn=get_data,
|
|
)
|
|
result = await resource.read()
|
|
assert isinstance(result, ResourceContent)
|
|
assert result.content == "async content"
|
|
assert result.meta == {"key": "value"}
|
|
|
|
|
|
class TestResourceContentToMcp:
|
|
"""Test ResourceContent.to_mcp_resource_contents method."""
|
|
|
|
def test_text_content_to_mcp(self):
|
|
"""Test converting text ResourceContent to MCP type."""
|
|
rc = ResourceContent(
|
|
content="hello world",
|
|
mime_type="text/html",
|
|
meta={"csp": "script-src 'self'"},
|
|
)
|
|
mcp_content = rc.to_mcp_resource_contents("resource://test")
|
|
|
|
assert hasattr(mcp_content, "text")
|
|
assert mcp_content.text == "hello world"
|
|
assert mcp_content.mimeType == "text/html"
|
|
assert mcp_content.meta == {"csp": "script-src 'self'"}
|
|
|
|
def test_binary_content_to_mcp(self):
|
|
"""Test converting binary ResourceContent to MCP type."""
|
|
rc = ResourceContent(
|
|
content=b"\x00\x01\x02",
|
|
mime_type="application/octet-stream",
|
|
meta={"encoding": "raw"},
|
|
)
|
|
mcp_content = rc.to_mcp_resource_contents("resource://test")
|
|
|
|
assert hasattr(mcp_content, "blob")
|
|
assert mcp_content.blob == "AAEC" # base64 of \x00\x01\x02
|
|
assert mcp_content.mimeType == "application/octet-stream"
|
|
assert mcp_content.meta == {"encoding": "raw"}
|
|
|
|
def test_default_mime_types(self):
|
|
"""Test default mime types are applied correctly."""
|
|
text_rc = ResourceContent(content="text")
|
|
text_mcp = text_rc.to_mcp_resource_contents("resource://test")
|
|
assert text_mcp.mimeType == "text/plain"
|
|
|
|
binary_rc = ResourceContent(content=b"binary")
|
|
binary_mcp = binary_rc.to_mcp_resource_contents("resource://test")
|
|
assert binary_mcp.mimeType == "application/octet-stream"
|
|
|
|
def test_none_meta(self):
|
|
"""Test that None meta is handled correctly."""
|
|
rc = ResourceContent(content="no meta")
|
|
mcp_content = rc.to_mcp_resource_contents("resource://test")
|
|
|
|
assert mcp_content.meta is None
|
|
|
|
|
|
class TestFunctionResourceCallable:
|
|
"""Test FunctionResource with callable objects."""
|
|
|
|
async def test_callable_object_sync(self):
|
|
"""Test that callable objects with sync __call__ work."""
|
|
|
|
class MyResource:
|
|
def __init__(self, value: str):
|
|
self.value = value
|
|
|
|
def __call__(self) -> str:
|
|
return f"value: {self.value}"
|
|
|
|
resource = FunctionResource.from_function(MyResource("test"), uri="fn://test")
|
|
result = await resource.read()
|
|
assert result == "value: test"
|
|
|
|
async def test_callable_object_async(self):
|
|
"""Test that callable objects with async __call__ work."""
|
|
|
|
class AsyncResource:
|
|
def __init__(self, value: str):
|
|
self.value = value
|
|
|
|
async def __call__(self) -> str:
|
|
return f"async value: {self.value}"
|
|
|
|
resource = FunctionResource.from_function(
|
|
AsyncResource("test"), uri="fn://test"
|
|
)
|
|
result = await resource.read()
|
|
assert result == "async value: test"
|
|
|
|
async def test_sync_resource_runs_concurrently(self):
|
|
"""Test that sync resources run in threadpool and don't block each other."""
|
|
import asyncio
|
|
import threading
|
|
|
|
num_calls = 3
|
|
barrier = threading.Barrier(num_calls, timeout=0.5)
|
|
|
|
def concurrent_resource() -> str:
|
|
barrier.wait()
|
|
return "done"
|
|
|
|
resource = FunctionResource.from_function(concurrent_resource, uri="fn://test")
|
|
|
|
# Run concurrent reads - will raise BrokenBarrierError if not concurrent
|
|
results = await asyncio.gather(
|
|
resource.read(),
|
|
resource.read(),
|
|
resource.read(),
|
|
)
|
|
assert results == ["done", "done", "done"]
|