From 09cf6550b5dab7208bf334d4637dbefac9fc879f Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 25 Apr 2025 17:49:42 -0400 Subject: [PATCH 1/8] Split server interaction tests into new file --- tests/server/test_server.py | 840 +--------------------- tests/server/test_server_interactions.py | 851 +++++++++++++++++++++++ 2 files changed, 853 insertions(+), 838 deletions(-) create mode 100644 tests/server/test_server_interactions.py diff --git a/tests/server/test_server.py b/tests/server/test_server.py index f27e0c042..7c4782eba 100644 --- a/tests/server/test_server.py +++ b/tests/server/test_server.py @@ -1,25 +1,11 @@ -import base64 -import json -from pathlib import Path -from typing import TYPE_CHECKING - import pytest from mcp.types import ( - BlobResourceContents, - ImageContent, TextContent, TextResourceContents, ) -from pydantic import AnyUrl, Field -from fastmcp import Client, Context, FastMCP -from fastmcp.exceptions import ClientError, NotFoundError, ToolError -from fastmcp.prompts.prompt import EmbeddedResource, Message, UserMessage -from fastmcp.resources import FileResource, FunctionResource -from fastmcp.utilities.types import Image - -if TYPE_CHECKING: - from fastmcp import Context +from fastmcp import Client, FastMCP +from fastmcp.exceptions import ClientError, NotFoundError class TestCreateServer: @@ -715,825 +701,3 @@ class TestPromptDecorator: assert len(prompts_dict) == 1 prompt = prompts_dict["sample_prompt"] assert prompt.tags == {"example", "test-tag"} - - -@pytest.fixture -def tool_server(): - mcp = FastMCP() - - @mcp.tool() - def add(x: int, y: int) -> int: - return x + y - - @mcp.tool() - def list_tool() -> list[str | int]: - return ["x", 2] - - @mcp.tool() - def error_tool() -> None: - raise ValueError("Test error") - - @mcp.tool() - def image_tool(path: str) -> Image: - return Image(path) - - @mcp.tool() - def mixed_content_tool() -> list[TextContent | ImageContent]: - return [ - TextContent(type="text", text="Hello"), - ImageContent(type="image", data="abc", mimeType="image/png"), - ] - - @mcp.tool() - def mixed_list_fn(image_path: str) -> list: - return [ - "text message", - Image(image_path), - {"key": "value"}, - TextContent(type="text", text="direct content"), - ] - - return mcp - - -class TestServerTools: - async def test_add_tool_exists(self, tool_server: FastMCP): - assert "add" in [t.name for t in await tool_server._mcp_list_tools()] - - async def test_list_tools(self, tool_server: FastMCP): - assert len(await tool_server._mcp_list_tools()) == 6 - - async def test_call_tool(self, tool_server: FastMCP): - result = await tool_server._mcp_call_tool("add", {"x": 1, "y": 2}) - assert isinstance(result[0], TextContent) - assert result[0].text == "3" - - async def test_call_tool_as_client(self, tool_server: FastMCP): - async with Client(tool_server) as client: - result = await client.call_tool("add", {"x": 1, "y": 2}) - assert isinstance(result[0], TextContent) - assert result[0].text == "3" - - async def test_call_tool_error(self, tool_server: FastMCP): - with pytest.raises(ToolError): - await tool_server._mcp_call_tool("error_tool", {}) - - async def test_call_tool_error_as_client(self, tool_server: FastMCP): - async with Client(tool_server) as client: - with pytest.raises(Exception): - await client.call_tool("error_tool", {}) - - async def test_call_tool_error_as_client_raw(self, tool_server: FastMCP): - async with Client(tool_server) as client: - result = await client.call_tool("error_tool", {}, _return_raw_result=True) - assert result.isError - assert isinstance(result.content[0], TextContent) - assert "Test error" in result.content[0].text - - async def test_tool_returns_list(self, tool_server: FastMCP): - result = await tool_server._mcp_call_tool("list_tool", {}) - assert isinstance(result[0], TextContent) - assert result[0].text == '["x", 2]' - - async def test_tool_image_helper(self, tool_server: FastMCP, tmp_path: Path): - # Create a test image - image_path = tmp_path / "test.png" - image_path.write_bytes(b"fake png data") - - result = await tool_server._mcp_call_tool( - "image_tool", {"path": str(image_path)} - ) - content = result[0] - assert isinstance(content, ImageContent) - assert content.type == "image" - assert content.mimeType == "image/png" - # Verify base64 encoding - decoded = base64.b64decode(content.data) - assert decoded == b"fake png data" - - async def test_tool_mixed_content(self, tool_server: FastMCP): - result = await tool_server._mcp_call_tool("mixed_content_tool", {}) - assert len(result) == 2 - content1 = result[0] - content2 = result[1] - assert isinstance(content1, TextContent) - assert content1.text == "Hello" - assert isinstance(content2, ImageContent) - assert content2.mimeType == "image/png" - assert content2.data == "abc" - - async def test_tool_mixed_list_with_image( - self, tool_server: FastMCP, tmp_path: Path - ): - """Test that lists containing Image objects and other types are handled - correctly. Note that the non-MCP content will be grouped together.""" - # Create a test image - image_path = tmp_path / "test.png" - image_path.write_bytes(b"test image data") - - result = await tool_server._mcp_call_tool( - "mixed_list_fn", {"image_path": str(image_path)} - ) - assert len(result) == 3 - # Check text conversion - content1 = result[0] - assert isinstance(content1, TextContent) - assert json.loads(content1.text) == ["text message", {"key": "value"}] - # Check image conversion - content2 = result[1] - assert isinstance(content2, ImageContent) - assert content2.mimeType == "image/png" - assert base64.b64decode(content2.data) == b"test image data" - # Check direct TextContent - content3 = result[2] - assert isinstance(content3, TextContent) - assert content3.text == "direct content" - - async def test_parameter_descriptions(self): - mcp = FastMCP("Test Server") - - @mcp.tool() - def greet( - name: str = Field(description="The name to greet"), - title: str = Field(description="Optional title", default=""), - ) -> str: - """A greeting tool""" - return f"Hello {title} {name}" - - tools = await mcp._mcp_list_tools() - assert len(tools) == 1 - tool = tools[0] - - # Check that parameter descriptions are present in the schema - properties = tool.inputSchema["properties"] - assert "name" in properties - assert properties["name"]["description"] == "The name to greet" - assert "title" in properties - assert properties["title"]["description"] == "Optional title" - - -class TestServerResources: - async def test_text_resource(self): - mcp = FastMCP() - - def get_text(): - return "Hello, world!" - - resource = FunctionResource( - uri=AnyUrl("resource://test"), name="test", fn=get_text - ) - mcp.add_resource(resource) - - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("resource://test")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Hello, world!" - - async def test_binary_resource(self): - mcp = FastMCP() - - def get_binary(): - return b"Binary data" - - resource = FunctionResource( - uri=AnyUrl("resource://binary"), - name="binary", - fn=get_binary, - mime_type="application/octet-stream", - ) - mcp.add_resource(resource) - - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("resource://binary")) - assert isinstance(result[0], BlobResourceContents) - assert result[0].blob == base64.b64encode(b"Binary data").decode() - - async def test_file_resource_text(self, tmp_path: Path): - mcp = FastMCP() - - # Create a text file - text_file = tmp_path / "test.txt" - text_file.write_text("Hello from file!") - - resource = FileResource( - uri=AnyUrl("file://test.txt"), name="test.txt", path=text_file - ) - mcp.add_resource(resource) - - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("file://test.txt")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Hello from file!" - - async def test_file_resource_binary(self, tmp_path: Path): - mcp = FastMCP() - - # Create a binary file - binary_file = tmp_path / "test.bin" - binary_file.write_bytes(b"Binary file data") - - resource = FileResource( - uri=AnyUrl("file://test.bin"), - name="test.bin", - path=binary_file, - mime_type="application/octet-stream", - ) - mcp.add_resource(resource) - - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("file://test.bin")) - assert isinstance(result[0], BlobResourceContents) - assert result[0].blob == base64.b64encode(b"Binary file data").decode() - - -class TestServerResourceTemplates: - async def test_resource_with_params_not_in_uri(self): - """Test that a resource with function parameters raises an error if the URI - parameters don't match""" - mcp = FastMCP() - - with pytest.raises( - ValueError, - match="URI template must contain at least one parameter", - ): - - @mcp.resource("resource://data") - def get_data_fn(param: str) -> str: - return f"Data: {param}" - - async def test_resource_with_uri_params_without_args(self): - """Test that a resource with URI parameters is automatically a template""" - mcp = FastMCP() - - with pytest.raises( - ValueError, - match="URI parameters .* must be a subset of the function arguments", - ): - - @mcp.resource("resource://{param}") - def get_data() -> str: - return "Data" - - async def test_resource_with_untyped_params(self): - """Test that a resource with untyped parameters raises an error""" - mcp = FastMCP() - - @mcp.resource("resource://{param}") - def get_data(param) -> str: - return "Data" - - async def test_resource_matching_params(self): - """Test that a resource with matching URI and function parameters works""" - mcp = FastMCP() - - @mcp.resource("resource://{name}/data") - def get_data(name: str) -> str: - return f"Data for {name}" - - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("resource://test/data")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Data for test" - - async def test_resource_mismatched_params(self): - """Test that mismatched parameters raise an error""" - mcp = FastMCP() - - with pytest.raises( - ValueError, - match="URI parameters .* must be a subset of the required function arguments", - ): - - @mcp.resource("resource://{name}/data") - def get_data(user: str) -> str: - return f"Data for {user}" - - async def test_resource_multiple_params(self): - """Test that multiple parameters work correctly""" - mcp = FastMCP() - - @mcp.resource("resource://{org}/{repo}/data") - def get_data(org: str, repo: str) -> str: - return f"Data for {org}/{repo}" - - async with Client(mcp) as client: - result = await client.read_resource( - AnyUrl("resource://cursor/fastmcp/data") - ) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Data for cursor/fastmcp" - - async def test_resource_multiple_mismatched_params(self): - """Test that mismatched parameters raise an error""" - mcp = FastMCP() - - with pytest.raises( - ValueError, - match="URI parameters .* must be a subset of the required function arguments", - ): - - @mcp.resource("resource://{org}/{repo}/data") - def get_data_mismatched(org: str, repo_2: str) -> str: - return f"Data for {org}" - - """Test that a resource with no parameters works as a regular resource""" - mcp = FastMCP() - - @mcp.resource("resource://static") - def get_static_data() -> str: - return "Static data" - - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("resource://static")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Static data" - - async def test_template_with_default_params(self): - """Test that a template can have default parameters.""" - mcp = FastMCP() - - @mcp.resource("math://add/{x}") - def add(x: int, y: int = 10) -> int: - return x + y - - # Verify it's registered as a template - templates_dict = await mcp.get_resource_templates() - templates = list(templates_dict.values()) - assert len(templates) == 1 - assert templates[0].uri_template == "math://add/{x}" - - # Call the template and verify it uses the default value - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("math://add/5")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "15" # 5 + default 10 - - # Can also call with explicit params - resource = await mcp._resource_manager.get_resource("math://add/7") - assert isinstance(resource, FunctionResource) - result = await resource.read() - assert result == "17" # 7 + default 10 - - async def test_template_to_resource_conversion(self): - """Test that a template can be converted to a resource.""" - mcp = FastMCP() - - @mcp.resource("resource://{name}/data") - def get_data(name: str) -> str: - return f"Data for {name}" - - # Verify it's registered as a template - templates_dict = await mcp.get_resource_templates() - templates = list(templates_dict.values()) - assert len(templates) == 1 - assert templates[0].uri_template == "resource://{name}/data" - - # When accessed, should create a concrete resource - resource = await mcp._resource_manager.get_resource("resource://test/data") - assert isinstance(resource, FunctionResource) - result = await resource.read() - assert result == "Data for test" - - async def test_stacked_resource_template_decorators(self): - """Test that resource template decorators can be stacked.""" - mcp = FastMCP() - - @mcp.resource("users://email/{email}") - @mcp.resource("users://name/{name}") - def lookup_user(name: str | None = None, email: str | None = None) -> dict: - if name: - return { - "id": "123", - "name": name, - "email": "dummy@example.com", - "lookup": "name", - } - elif email: - return { - "id": "123", - "name": "Test User", - "email": email, - "lookup": "email", - } - else: - raise ValueError("Either name or email must be provided") - - # Verify both templates are registered - templates_dict = await mcp.get_resource_templates() - templates = list(templates_dict.values()) - assert len(templates) == 2 - template_uris = {t.uri_template for t in templates} - assert "users://email/{email}" in template_uris - assert "users://name/{name}" in template_uris - - # Test lookup by email - async with Client(mcp) as client: - email_result = await client.read_resource( - AnyUrl("users://email/user@example.com") - ) - assert isinstance(email_result[0], TextResourceContents) - email_data = json.loads(email_result[0].text) - assert email_data["lookup"] == "email" - assert email_data["email"] == "user@example.com" - - # Test lookup by name - name_result = await client.read_resource(AnyUrl("users://name/John")) - assert isinstance(name_result[0], TextResourceContents) - name_data = json.loads(name_result[0].text) - assert name_data["lookup"] == "name" - assert name_data["name"] == "John" - assert name_data["email"] == "dummy@example.com" - - async def test_template_decorator_with_tags(self): - mcp = FastMCP() - - @mcp.resource("resource://{param}", tags={"template", "test-tag"}) - def template_resource(param: str) -> str: - return f"Template resource: {param}" - - templates_dict = await mcp.get_resource_templates() - template = templates_dict["resource://{param}"] - assert template.tags == {"template", "test-tag"} - - async def test_template_decorator_wildcard_param(self): - mcp = FastMCP() - - @mcp.resource("resource://{param*}") - def template_resource(param: str) -> str: - return f"Template resource: {param}" - - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("resource://test/data")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Template resource: test/data" - - async def test_templates_match_in_order_of_definition(self): - """ - If a wildcard template is defined first, it will take priority over another - matching template. - - """ - mcp = FastMCP() - - @mcp.resource("resource://{param*}") - def template_resource(param: str) -> str: - return f"Template resource 1: {param}" - - @mcp.resource("resource://{x}/{y}") - def template_resource_with_params(x: str, y: str) -> str: - return f"Template resource 2: {x}/{y}" - - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("resource://a/b/c")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Template resource 1: a/b/c" - - result = await client.read_resource(AnyUrl("resource://a/b")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Template resource 1: a/b" - - async def test_templates_shadow_each_other_reorder(self): - """ - If a wildcard template is defined second, it will *not* take priority over - another matching template. - """ - mcp = FastMCP() - - @mcp.resource("resource://{x}/{y}") - def template_resource_with_params(x: str, y: str) -> str: - return f"Template resource 1: {x}/{y}" - - @mcp.resource("resource://{param*}") - def template_resource(param: str) -> str: - return f"Template resource 2: {param}" - - async with Client(mcp) as client: - result = await client.read_resource(AnyUrl("resource://a/b/c")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Template resource 2: a/b/c" - - result = await client.read_resource(AnyUrl("resource://a/b")) - assert isinstance(result[0], TextResourceContents) - assert result[0].text == "Template resource 1: a/b" - - -class TestContextInjection: - """Test context injection in tools.""" - - async def test_context_detection(self): - """Test that context parameters are properly detected.""" - mcp = FastMCP() - - def tool_with_context(x: int, ctx: Context) -> str: - return f"Request {ctx.request_id}: {x}" - - tool = mcp._tool_manager.add_tool_from_fn(tool_with_context) - assert tool.context_kwarg == "ctx" - - async def test_context_injection(self): - """Test that context is properly injected into tool calls.""" - mcp = FastMCP() - - def tool_with_context(x: int, ctx: Context) -> str: - assert ctx.request_id is not None - return f"Request {ctx.request_id}: {x}" - - mcp.add_tool(tool_with_context) - async with Client(mcp) as client: - result = await client.call_tool("tool_with_context", {"x": 42}) - assert len(result) == 1 - content = result[0] - assert isinstance(content, TextContent) - assert "Request" in content.text - assert "42" in content.text - - async def test_async_context(self): - """Test that context works in async functions.""" - mcp = FastMCP() - - async def async_tool(x: int, ctx: Context) -> str: - assert ctx.request_id is not None - return f"Async request {ctx.request_id}: {x}" - - mcp.add_tool(async_tool) - async with Client(mcp) as client: - result = await client.call_tool("async_tool", {"x": 42}) - assert len(result) == 1 - content = result[0] - assert isinstance(content, TextContent) - assert "Async request" in content.text - assert "42" in content.text - - async def test_context_logging(self): - from unittest.mock import patch - - import mcp.server.session - - """Test that context logging methods work.""" - mcp = FastMCP() - - async def logging_tool(msg: str, ctx: Context) -> str: - await ctx.debug("Debug message") - await ctx.info("Info message") - await ctx.warning("Warning message") - await ctx.error("Error message") - return f"Logged messages for {msg}" - - mcp.add_tool(logging_tool) - - with patch("mcp.server.session.ServerSession.send_log_message") as mock_log: - async with Client(mcp) as client: - result = await client.call_tool("logging_tool", {"msg": "test"}) - assert len(result) == 1 - content = result[0] - assert isinstance(content, TextContent) - assert "Logged messages for test" in content.text - - assert mock_log.call_count == 4 - mock_log.assert_any_call( - level="debug", data="Debug message", logger=None - ) - mock_log.assert_any_call(level="info", data="Info message", logger=None) - mock_log.assert_any_call( - level="warning", data="Warning message", logger=None - ) - mock_log.assert_any_call( - level="error", data="Error message", logger=None - ) - - async def test_optional_context(self): - """Test that context is optional.""" - mcp = FastMCP() - - def no_context(x: int) -> int: - return x * 2 - - mcp.add_tool(no_context) - async with Client(mcp) as client: - result = await client.call_tool("no_context", {"x": 21}) - assert len(result) == 1 - content = result[0] - assert isinstance(content, TextContent) - assert content.text == "42" - - async def test_context_resource_access(self): - """Test that context can access resources.""" - mcp = FastMCP() - - @mcp.resource("test://data") - def test_resource() -> str: - return "resource data" - - @mcp.tool() - async def tool_with_resource(ctx: Context) -> str: - r_iter = await ctx.read_resource("test://data") - r_list = list(r_iter) - assert len(r_list) == 1 - r = r_list[0] - return f"Read resource: {r.content} with mime type {r.mime_type}" - - async with Client(mcp) as client: - result = await client.call_tool("tool_with_resource", {}) - assert len(result) == 1 - content = result[0] - assert isinstance(content, TextContent) - assert "Read resource: resource data" in content.text - - -class TestServerPrompts: - """Test prompt functionality in FastMCP server.""" - - async def test_prompt_decorator(self): - """Test that the prompt decorator registers prompts correctly.""" - mcp = FastMCP() - - @mcp.prompt() - def fn() -> str: - return "Hello, world!" - - prompts_dict = await mcp.get_prompts() - assert len(prompts_dict) == 1 - prompt = prompts_dict["fn"] - assert prompt.name == "fn" - # Don't compare functions directly since validate_call wraps them - content = await prompt.render() - assert isinstance(content[0].content, TextContent) - assert content[0].content.text == "Hello, world!" - - async def test_prompt_decorator_with_name(self): - """Test prompt decorator with custom name.""" - mcp = FastMCP() - - @mcp.prompt(name="custom_name") - def fn() -> str: - return "Hello, world!" - - prompts_dict = await mcp.get_prompts() - assert len(prompts_dict) == 1 - prompt = prompts_dict["custom_name"] - assert prompt.name == "custom_name" - content = await prompt.render() - assert isinstance(content[0].content, TextContent) - assert content[0].content.text == "Hello, world!" - - async def test_prompt_decorator_with_description(self): - """Test prompt decorator with custom description.""" - mcp = FastMCP() - - @mcp.prompt(description="A custom description") - def fn() -> str: - return "Hello, world!" - - prompts_dict = await mcp.get_prompts() - assert len(prompts_dict) == 1 - prompt = prompts_dict["fn"] - assert prompt.description == "A custom description" - content = await prompt.render() - assert isinstance(content[0].content, TextContent) - assert content[0].content.text == "Hello, world!" - - def test_prompt_decorator_error(self): - """Test error when decorator is used incorrectly.""" - mcp = FastMCP() - with pytest.raises(TypeError, match="decorator was used incorrectly"): - - @mcp.prompt # type: ignore - def fn() -> str: - return "Hello, world!" - - async def test_list_prompts(self): - """Test listing prompts through MCP protocol.""" - mcp = FastMCP() - - @mcp.prompt() - def fn(name: str, optional: str = "default") -> str: - return f"Hello, {name}! {optional}" - - prompts_dict = await mcp.get_prompts() - assert len(prompts_dict) == 1 - - async with Client(mcp) as client: - prompts = await client.list_prompts() - assert len(prompts) == 1 - assert prompts[0].name == "fn" - assert prompts[0].description is None - assert prompts[0].arguments is not None - assert len(prompts[0].arguments) == 2 - assert prompts[0].arguments[0].name == "name" - assert prompts[0].arguments[0].required is True - assert prompts[0].arguments[1].name == "optional" - assert prompts[0].arguments[1].required is False - - async def test_get_prompt(self): - """Test getting a prompt through MCP protocol.""" - mcp = FastMCP() - - @mcp.prompt() - def fn(name: str) -> str: - return f"Hello, {name}!" - - async with Client(mcp) as client: - result = await client.get_prompt("fn", {"name": "World"}) - assert len(result) == 1 - message = result[0] - assert message.role == "user" - content = message.content - assert isinstance(content, TextContent) - assert content.text == "Hello, World!" - - async def test_get_prompt_with_resource(self): - """Test getting a prompt that returns resource content.""" - mcp = FastMCP() - - @mcp.prompt() - def fn() -> Message: - return UserMessage( - content=EmbeddedResource( - type="resource", - resource=TextResourceContents( - uri=AnyUrl("file://file.txt"), - text="File contents", - mimeType="text/plain", - ), - ) - ) - - async with Client(mcp) as client: - result = await client.get_prompt("fn") - assert result[0].role == "user" - content = result[0].content - assert isinstance(content, EmbeddedResource) - resource = content.resource - assert isinstance(resource, TextResourceContents) - assert resource.text == "File contents" - assert resource.mimeType == "text/plain" - - async def test_get_unknown_prompt(self): - """Test error when getting unknown prompt.""" - mcp = FastMCP() - with pytest.raises(ClientError, match="Unknown prompt"): - async with Client(mcp) as client: - await client.get_prompt("unknown") - - async def test_get_prompt_missing_args(self): - """Test error when required arguments are missing.""" - mcp = FastMCP() - - @mcp.prompt() - def prompt_fn(name: str) -> str: - return f"Hello, {name}!" - - with pytest.raises(ClientError, match="Missing required arguments"): - async with Client(mcp) as client: - await client.get_prompt("prompt_fn") - - async def test_tool_decorator_with_tags(self): - """Test that the tool decorator properly sets tags.""" - mcp = FastMCP() - - @mcp.tool(tags={"example", "test-tag"}) - def sample_tool(x: int) -> int: - return x * 2 - - # Verify the tags were set correctly - tools = mcp._tool_manager.list_tools() - assert len(tools) == 1 - assert tools[0].tags == {"example", "test-tag"} - - async def test_resource_decorator_with_tags(self): - """Test that the resource decorator supports tags.""" - mcp = FastMCP() - - @mcp.resource("resource://data", tags={"example", "test-tag"}) - def get_data() -> str: - return "Hello, world!" - - resources_dict = await mcp.get_resources() - resources = list(resources_dict.values()) - assert len(resources) == 1 - assert resources[0].tags == {"example", "test-tag"} - - async def test_template_decorator_with_tags(self): - """Test that the template decorator properly sets tags.""" - mcp = FastMCP() - - @mcp.resource("resource://{param}", tags={"template", "test-tag"}) - def template_resource(param: str) -> str: - return f"Template resource: {param}" - - templates_dict = await mcp.get_resource_templates() - template = templates_dict["resource://{param}"] - assert template.tags == {"template", "test-tag"} - - async def test_prompt_decorator_with_tags(self): - """Test that the prompt decorator properly sets tags.""" - mcp = FastMCP() - - @mcp.prompt(tags={"example", "test-tag"}) - def sample_prompt() -> str: - return "Hello, world!" - - prompts_dict = await mcp.get_prompts() - assert len(prompts_dict) == 1 - prompt = prompts_dict["sample_prompt"] - assert prompt.tags == {"example", "test-tag"} diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py new file mode 100644 index 000000000..6ac501787 --- /dev/null +++ b/tests/server/test_server_interactions.py @@ -0,0 +1,851 @@ +import base64 +import json +from pathlib import Path + +import pytest +from mcp.types import ( + BlobResourceContents, + ImageContent, + TextContent, + TextResourceContents, +) +from pydantic import AnyUrl, Field + +from fastmcp import Client, Context, FastMCP +from fastmcp.exceptions import ClientError +from fastmcp.prompts.prompt import EmbeddedResource, Message, UserMessage +from fastmcp.resources import FileResource, FunctionResource +from fastmcp.utilities.types import Image + + +@pytest.fixture +def tool_server(): + mcp = FastMCP() + + @mcp.tool() + def add(x: int, y: int) -> int: + return x + y + + @mcp.tool() + def list_tool() -> list[str | int]: + return ["x", 2] + + @mcp.tool() + def error_tool() -> None: + raise ValueError("Test error") + + @mcp.tool() + def image_tool(path: str) -> Image: + return Image(path) + + @mcp.tool() + def mixed_content_tool() -> list[TextContent | ImageContent]: + return [ + TextContent(type="text", text="Hello"), + ImageContent(type="image", data="abc", mimeType="image/png"), + ] + + @mcp.tool() + def mixed_list_fn(image_path: str) -> list: + return [ + "text message", + Image(image_path), + {"key": "value"}, + TextContent(type="text", text="direct content"), + ] + + return mcp + + +class TestTools: + async def test_add_tool_exists(self, tool_server: FastMCP): + async with Client(tool_server) as client: + tools = await client.list_tools() + assert "add" in [t.name for t in tools] + + async def test_list_tools(self, tool_server: FastMCP): + async with Client(tool_server) as client: + assert len(await client.list_tools()) == 6 + + async def test_call_tool(self, tool_server: FastMCP): + async with Client(tool_server) as client: + result = await client.call_tool("add", {"x": 1, "y": 2}) + assert isinstance(result[0], TextContent) + assert result[0].text == "3" + + async def test_call_tool_as_client(self, tool_server: FastMCP): + async with Client(tool_server) as client: + result = await client.call_tool("add", {"x": 1, "y": 2}) + assert isinstance(result[0], TextContent) + assert result[0].text == "3" + + async def test_call_tool_error(self, tool_server: FastMCP): + async with Client(tool_server) as client: + with pytest.raises(Exception): + await client.call_tool("error_tool", {}) + + async def test_call_tool_error_as_client(self, tool_server: FastMCP): + async with Client(tool_server) as client: + with pytest.raises(Exception): + await client.call_tool("error_tool", {}) + + async def test_call_tool_error_as_client_raw(self, tool_server: FastMCP): + async with Client(tool_server) as client: + result = await client.call_tool("error_tool", {}, _return_raw_result=True) + assert result.isError + assert isinstance(result.content[0], TextContent) + assert "Test error" in result.content[0].text + + async def test_tool_returns_list(self, tool_server: FastMCP): + async with Client(tool_server) as client: + result = await client.call_tool("list_tool", {}) + assert isinstance(result[0], TextContent) + assert result[0].text == '["x", 2]' + + async def test_tool_image_helper(self, tool_server: FastMCP, tmp_path: Path): + # Create a test image + image_path = tmp_path / "test.png" + image_path.write_bytes(b"fake png data") + + async with Client(tool_server) as client: + result = await client.call_tool("image_tool", {"path": str(image_path)}) + content = result[0] + assert isinstance(content, ImageContent) + assert content.type == "image" + assert content.mimeType == "image/png" + # Verify base64 encoding + decoded = base64.b64decode(content.data) + assert decoded == b"fake png data" + + async def test_tool_mixed_content(self, tool_server: FastMCP): + async with Client(tool_server) as client: + result = await client.call_tool("mixed_content_tool", {}) + assert len(result) == 2 + content1 = result[0] + content2 = result[1] + assert isinstance(content1, TextContent) + assert content1.text == "Hello" + assert isinstance(content2, ImageContent) + assert content2.mimeType == "image/png" + assert content2.data == "abc" + + async def test_tool_mixed_list_with_image( + self, tool_server: FastMCP, tmp_path: Path + ): + """Test that lists containing Image objects and other types are handled + correctly. Note that the non-MCP content will be grouped together.""" + # Create a test image + image_path = tmp_path / "test.png" + image_path.write_bytes(b"test image data") + + async with Client(tool_server) as client: + result = await client.call_tool( + "mixed_list_fn", {"image_path": str(image_path)} + ) + assert len(result) == 3 + # Check text conversion + content1 = result[0] + assert isinstance(content1, TextContent) + assert json.loads(content1.text) == ["text message", {"key": "value"}] + # Check image conversion + content2 = result[1] + assert isinstance(content2, ImageContent) + assert content2.mimeType == "image/png" + assert base64.b64decode(content2.data) == b"test image data" + # Check direct TextContent + content3 = result[2] + assert isinstance(content3, TextContent) + assert content3.text == "direct content" + + async def test_parameter_descriptions(self): + mcp = FastMCP("Test Server") + + @mcp.tool() + def greet( + name: str = Field(description="The name to greet"), + title: str = Field(description="Optional title", default=""), + ) -> str: + """A greeting tool""" + return f"Hello {title} {name}" + + async with Client(mcp) as client: + tools = await client.list_tools() + assert len(tools) == 1 + tool = tools[0] + + # Check that parameter descriptions are present in the schema + properties = tool.inputSchema["properties"] + assert "name" in properties + assert properties["name"]["description"] == "The name to greet" + assert "title" in properties + assert properties["title"]["description"] == "Optional title" + + +class TestResources: + async def test_text_resource(self): + mcp = FastMCP() + + def get_text(): + return "Hello, world!" + + resource = FunctionResource( + uri=AnyUrl("resource://test"), name="test", fn=get_text + ) + mcp.add_resource(resource) + + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("resource://test")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Hello, world!" + + async def test_binary_resource(self): + mcp = FastMCP() + + def get_binary(): + return b"Binary data" + + resource = FunctionResource( + uri=AnyUrl("resource://binary"), + name="binary", + fn=get_binary, + mime_type="application/octet-stream", + ) + mcp.add_resource(resource) + + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("resource://binary")) + assert isinstance(result[0], BlobResourceContents) + assert result[0].blob == base64.b64encode(b"Binary data").decode() + + async def test_file_resource_text(self, tmp_path: Path): + mcp = FastMCP() + + # Create a text file + text_file = tmp_path / "test.txt" + text_file.write_text("Hello from file!") + + resource = FileResource( + uri=AnyUrl("file://test.txt"), name="test.txt", path=text_file + ) + mcp.add_resource(resource) + + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("file://test.txt")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Hello from file!" + + async def test_file_resource_binary(self, tmp_path: Path): + mcp = FastMCP() + + # Create a binary file + binary_file = tmp_path / "test.bin" + binary_file.write_bytes(b"Binary file data") + + resource = FileResource( + uri=AnyUrl("file://test.bin"), + name="test.bin", + path=binary_file, + mime_type="application/octet-stream", + ) + mcp.add_resource(resource) + + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("file://test.bin")) + assert isinstance(result[0], BlobResourceContents) + assert result[0].blob == base64.b64encode(b"Binary file data").decode() + + +class TestResourceTemplates: + async def test_resource_with_params_not_in_uri(self): + """Test that a resource with function parameters raises an error if the URI + parameters don't match""" + mcp = FastMCP() + + with pytest.raises( + ValueError, + match="URI template must contain at least one parameter", + ): + + @mcp.resource("resource://data") + def get_data_fn(param: str) -> str: + return f"Data: {param}" + + async def test_resource_with_uri_params_without_args(self): + """Test that a resource with URI parameters is automatically a template""" + mcp = FastMCP() + + with pytest.raises( + ValueError, + match="URI parameters .* must be a subset of the function arguments", + ): + + @mcp.resource("resource://{param}") + def get_data() -> str: + return "Data" + + async def test_resource_with_untyped_params(self): + """Test that a resource with untyped parameters raises an error""" + mcp = FastMCP() + + @mcp.resource("resource://{param}") + def get_data(param) -> str: + return "Data" + + async def test_resource_matching_params(self): + """Test that a resource with matching URI and function parameters works""" + mcp = FastMCP() + + @mcp.resource("resource://{name}/data") + def get_data(name: str) -> str: + return f"Data for {name}" + + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("resource://test/data")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Data for test" + + async def test_resource_mismatched_params(self): + """Test that mismatched parameters raise an error""" + mcp = FastMCP() + + with pytest.raises( + ValueError, + match="URI parameters .* must be a subset of the required function arguments", + ): + + @mcp.resource("resource://{name}/data") + def get_data(user: str) -> str: + return f"Data for {user}" + + async def test_resource_multiple_params(self): + """Test that multiple parameters work correctly""" + mcp = FastMCP() + + @mcp.resource("resource://{org}/{repo}/data") + def get_data(org: str, repo: str) -> str: + return f"Data for {org}/{repo}" + + async with Client(mcp) as client: + result = await client.read_resource( + AnyUrl("resource://cursor/fastmcp/data") + ) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Data for cursor/fastmcp" + + async def test_resource_multiple_mismatched_params(self): + """Test that mismatched parameters raise an error""" + mcp = FastMCP() + + with pytest.raises( + ValueError, + match="URI parameters .* must be a subset of the required function arguments", + ): + + @mcp.resource("resource://{org}/{repo}/data") + def get_data_mismatched(org: str, repo_2: str) -> str: + return f"Data for {org}" + + """Test that a resource with no parameters works as a regular resource""" + mcp = FastMCP() + + @mcp.resource("resource://static") + def get_static_data() -> str: + return "Static data" + + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("resource://static")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Static data" + + async def test_template_with_default_params(self): + """Test that a template can have default parameters.""" + mcp = FastMCP() + + @mcp.resource("math://add/{x}") + def add(x: int, y: int = 10) -> int: + return x + y + + # Verify it's registered as a template + templates_dict = await mcp.get_resource_templates() + templates = list(templates_dict.values()) + assert len(templates) == 1 + assert templates[0].uri_template == "math://add/{x}" + + # Call the template and verify it uses the default value + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("math://add/5")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "15" # 5 + default 10 + + # Can also call with explicit params + result2 = await client.read_resource(AnyUrl("math://add/7")) + assert isinstance(result2[0], TextResourceContents) + assert result2[0].text == "17" # 7 + default 10 + + async def test_template_to_resource_conversion(self): + """Test that a template can be converted to a resource.""" + mcp = FastMCP() + + @mcp.resource("resource://{name}/data") + def get_data(name: str) -> str: + return f"Data for {name}" + + # Verify it's registered as a template + templates_dict = await mcp.get_resource_templates() + templates = list(templates_dict.values()) + assert len(templates) == 1 + assert templates[0].uri_template == "resource://{name}/data" + + # When accessed, should create a concrete resource + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("resource://test/data")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Data for test" + + async def test_stacked_resource_template_decorators(self): + """Test that resource template decorators can be stacked.""" + mcp = FastMCP() + + @mcp.resource("users://email/{email}") + @mcp.resource("users://name/{name}") + def lookup_user(name: str | None = None, email: str | None = None) -> dict: + if name: + return { + "id": "123", + "name": name, + "email": "dummy@example.com", + "lookup": "name", + } + elif email: + return { + "id": "123", + "name": "Test User", + "email": email, + "lookup": "email", + } + else: + raise ValueError("Either name or email must be provided") + + # Verify both templates are registered + templates_dict = await mcp.get_resource_templates() + templates = list(templates_dict.values()) + assert len(templates) == 2 + template_uris = {t.uri_template for t in templates} + assert "users://email/{email}" in template_uris + assert "users://name/{name}" in template_uris + + # Test lookup by email + async with Client(mcp) as client: + email_result = await client.read_resource( + AnyUrl("users://email/user@example.com") + ) + assert isinstance(email_result[0], TextResourceContents) + email_data = json.loads(email_result[0].text) + assert email_data["lookup"] == "email" + assert email_data["email"] == "user@example.com" + + # Test lookup by name + name_result = await client.read_resource(AnyUrl("users://name/John")) + assert isinstance(name_result[0], TextResourceContents) + name_data = json.loads(name_result[0].text) + assert name_data["lookup"] == "name" + assert name_data["name"] == "John" + assert name_data["email"] == "dummy@example.com" + + async def test_template_decorator_with_tags(self): + mcp = FastMCP() + + @mcp.resource("resource://{param}", tags={"template", "test-tag"}) + def template_resource(param: str) -> str: + return f"Template resource: {param}" + + templates_dict = await mcp.get_resource_templates() + template = templates_dict["resource://{param}"] + assert template.tags == {"template", "test-tag"} + + async def test_template_decorator_wildcard_param(self): + mcp = FastMCP() + + @mcp.resource("resource://{param*}") + def template_resource(param: str) -> str: + return f"Template resource: {param}" + + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("resource://test/data")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Template resource: test/data" + + async def test_templates_match_in_order_of_definition(self): + """ + If a wildcard template is defined first, it will take priority over another + matching template. + + """ + mcp = FastMCP() + + @mcp.resource("resource://{param*}") + def template_resource(param: str) -> str: + return f"Template resource 1: {param}" + + @mcp.resource("resource://{x}/{y}") + def template_resource_with_params(x: str, y: str) -> str: + return f"Template resource 2: {x}/{y}" + + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("resource://a/b/c")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Template resource 1: a/b/c" + + result = await client.read_resource(AnyUrl("resource://a/b")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Template resource 1: a/b" + + async def test_templates_shadow_each_other_reorder(self): + """ + If a wildcard template is defined second, it will *not* take priority over + another matching template. + """ + mcp = FastMCP() + + @mcp.resource("resource://{x}/{y}") + def template_resource_with_params(x: str, y: str) -> str: + return f"Template resource 1: {x}/{y}" + + @mcp.resource("resource://{param*}") + def template_resource(param: str) -> str: + return f"Template resource 2: {param}" + + async with Client(mcp) as client: + result = await client.read_resource(AnyUrl("resource://a/b/c")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Template resource 2: a/b/c" + + result = await client.read_resource(AnyUrl("resource://a/b")) + assert isinstance(result[0], TextResourceContents) + assert result[0].text == "Template resource 1: a/b" + + +class TestContextInjection: + """Test context injection in tools.""" + + async def test_context_detection(self): + """Test that context parameters are properly detected.""" + mcp = FastMCP() + + def tool_with_context(x: int, ctx: Context) -> str: + return f"Request {ctx.request_id}: {x}" + + mcp.add_tool(tool_with_context) + async with Client(mcp) as client: + tools = await client.list_tools() + assert len(tools) == 1 + assert tools[0].name == "tool_with_context" + + async def test_context_injection(self): + """Test that context is properly injected into tool calls.""" + mcp = FastMCP() + + def tool_with_context(x: int, ctx: Context) -> str: + assert ctx.request_id is not None + return f"Request {ctx.request_id}: {x}" + + mcp.add_tool(tool_with_context) + async with Client(mcp) as client: + result = await client.call_tool("tool_with_context", {"x": 42}) + assert len(result) == 1 + content = result[0] + assert isinstance(content, TextContent) + assert "Request" in content.text + assert "42" in content.text + + async def test_async_context(self): + """Test that context works in async functions.""" + mcp = FastMCP() + + async def async_tool(x: int, ctx: Context) -> str: + assert ctx.request_id is not None + return f"Async request {ctx.request_id}: {x}" + + mcp.add_tool(async_tool) + async with Client(mcp) as client: + result = await client.call_tool("async_tool", {"x": 42}) + assert len(result) == 1 + content = result[0] + assert isinstance(content, TextContent) + assert "Async request" in content.text + assert "42" in content.text + + async def test_context_logging(self): + from unittest.mock import patch + + import mcp.server.session + + """Test that context logging methods work.""" + mcp = FastMCP() + + async def logging_tool(msg: str, ctx: Context) -> str: + await ctx.debug("Debug message") + await ctx.info("Info message") + await ctx.warning("Warning message") + await ctx.error("Error message") + return f"Logged messages for {msg}" + + mcp.add_tool(logging_tool) + + with patch("mcp.server.session.ServerSession.send_log_message") as mock_log: + async with Client(mcp) as client: + result = await client.call_tool("logging_tool", {"msg": "test"}) + assert len(result) == 1 + content = result[0] + assert isinstance(content, TextContent) + assert "Logged messages for test" in content.text + + assert mock_log.call_count == 4 + mock_log.assert_any_call( + level="debug", data="Debug message", logger=None + ) + mock_log.assert_any_call(level="info", data="Info message", logger=None) + mock_log.assert_any_call( + level="warning", data="Warning message", logger=None + ) + mock_log.assert_any_call( + level="error", data="Error message", logger=None + ) + + async def test_optional_context(self): + """Test that context is optional.""" + mcp = FastMCP() + + def no_context(x: int) -> int: + return x * 2 + + mcp.add_tool(no_context) + async with Client(mcp) as client: + result = await client.call_tool("no_context", {"x": 21}) + assert len(result) == 1 + content = result[0] + assert isinstance(content, TextContent) + assert content.text == "42" + + async def test_context_resource_access(self): + """Test that context can access resources.""" + mcp = FastMCP() + + @mcp.resource("test://data") + def test_resource() -> str: + return "resource data" + + @mcp.tool() + async def tool_with_resource(ctx: Context) -> str: + r_iter = await ctx.read_resource("test://data") + r_list = list(r_iter) + assert len(r_list) == 1 + r = r_list[0] + return f"Read resource: {r.content} with mime type {r.mime_type}" + + async with Client(mcp) as client: + result = await client.call_tool("tool_with_resource", {}) + assert len(result) == 1 + content = result[0] + assert isinstance(content, TextContent) + assert "Read resource: resource data" in content.text + + async def test_tool_decorator_with_tags(self): + """Test that the tool decorator properly sets tags.""" + mcp = FastMCP() + + @mcp.tool(tags={"example", "test-tag"}) + def sample_tool(x: int) -> int: + return x * 2 + + # Verify the tool exists + async with Client(mcp) as client: + tools = await client.list_tools() + assert len(tools) == 1 + # Note: MCPTool from the client API doesn't expose tags + + +class TestPrompts: + """Test prompt functionality in FastMCP server.""" + + async def test_prompt_decorator(self): + """Test that the prompt decorator registers prompts correctly.""" + mcp = FastMCP() + + @mcp.prompt() + def fn() -> str: + return "Hello, world!" + + prompts_dict = await mcp.get_prompts() + assert len(prompts_dict) == 1 + prompt = prompts_dict["fn"] + assert prompt.name == "fn" + # Don't compare functions directly since validate_call wraps them + content = await prompt.render() + assert isinstance(content[0].content, TextContent) + assert content[0].content.text == "Hello, world!" + + async def test_prompt_decorator_with_name(self): + """Test prompt decorator with custom name.""" + mcp = FastMCP() + + @mcp.prompt(name="custom_name") + def fn() -> str: + return "Hello, world!" + + prompts_dict = await mcp.get_prompts() + assert len(prompts_dict) == 1 + prompt = prompts_dict["custom_name"] + assert prompt.name == "custom_name" + content = await prompt.render() + assert isinstance(content[0].content, TextContent) + assert content[0].content.text == "Hello, world!" + + async def test_prompt_decorator_with_description(self): + """Test prompt decorator with custom description.""" + mcp = FastMCP() + + @mcp.prompt(description="A custom description") + def fn() -> str: + return "Hello, world!" + + prompts_dict = await mcp.get_prompts() + assert len(prompts_dict) == 1 + prompt = prompts_dict["fn"] + assert prompt.description == "A custom description" + content = await prompt.render() + assert isinstance(content[0].content, TextContent) + assert content[0].content.text == "Hello, world!" + + def test_prompt_decorator_error(self): + """Test error when decorator is used incorrectly.""" + mcp = FastMCP() + with pytest.raises(TypeError, match="decorator was used incorrectly"): + + @mcp.prompt # type: ignore + def fn() -> str: + return "Hello, world!" + + async def test_list_prompts(self): + """Test listing prompts through MCP protocol.""" + mcp = FastMCP() + + @mcp.prompt() + def fn(name: str, optional: str = "default") -> str: + return f"Hello, {name}! {optional}" + + prompts_dict = await mcp.get_prompts() + assert len(prompts_dict) == 1 + + async with Client(mcp) as client: + prompts = await client.list_prompts() + assert len(prompts) == 1 + assert prompts[0].name == "fn" + assert prompts[0].description is None + assert prompts[0].arguments is not None + assert len(prompts[0].arguments) == 2 + assert prompts[0].arguments[0].name == "name" + assert prompts[0].arguments[0].required is True + assert prompts[0].arguments[1].name == "optional" + assert prompts[0].arguments[1].required is False + + async def test_get_prompt(self): + """Test getting a prompt through MCP protocol.""" + mcp = FastMCP() + + @mcp.prompt() + def fn(name: str) -> str: + return f"Hello, {name}!" + + async with Client(mcp) as client: + result = await client.get_prompt("fn", {"name": "World"}) + assert len(result) == 1 + message = result[0] + assert message.role == "user" + content = message.content + assert isinstance(content, TextContent) + assert content.text == "Hello, World!" + + async def test_get_prompt_with_resource(self): + """Test getting a prompt that returns resource content.""" + mcp = FastMCP() + + @mcp.prompt() + def fn() -> Message: + return UserMessage( + content=EmbeddedResource( + type="resource", + resource=TextResourceContents( + uri=AnyUrl("file://file.txt"), + text="File contents", + mimeType="text/plain", + ), + ) + ) + + async with Client(mcp) as client: + result = await client.get_prompt("fn") + assert result[0].role == "user" + content = result[0].content + assert isinstance(content, EmbeddedResource) + resource = content.resource + assert isinstance(resource, TextResourceContents) + assert resource.text == "File contents" + assert resource.mimeType == "text/plain" + + async def test_get_unknown_prompt(self): + """Test error when getting unknown prompt.""" + mcp = FastMCP() + with pytest.raises(ClientError, match="Unknown prompt"): + async with Client(mcp) as client: + await client.get_prompt("unknown") + + async def test_get_prompt_missing_args(self): + """Test error when required arguments are missing.""" + mcp = FastMCP() + + @mcp.prompt() + def prompt_fn(name: str) -> str: + return f"Hello, {name}!" + + with pytest.raises(ClientError, match="Missing required arguments"): + async with Client(mcp) as client: + await client.get_prompt("prompt_fn") + + async def test_resource_decorator_with_tags(self): + """Test that the resource decorator supports tags.""" + mcp = FastMCP() + + @mcp.resource("resource://data", tags={"example", "test-tag"}) + def get_data() -> str: + return "Hello, world!" + + resources_dict = await mcp.get_resources() + resources = list(resources_dict.values()) + assert len(resources) == 1 + assert resources[0].tags == {"example", "test-tag"} + + async def test_template_decorator_with_tags(self): + """Test that the template decorator properly sets tags.""" + mcp = FastMCP() + + @mcp.resource("resource://{param}", tags={"template", "test-tag"}) + def template_resource(param: str) -> str: + return f"Template resource: {param}" + + templates_dict = await mcp.get_resource_templates() + template = templates_dict["resource://{param}"] + assert template.tags == {"template", "test-tag"} + + async def test_prompt_decorator_with_tags(self): + """Test that the prompt decorator properly sets tags.""" + mcp = FastMCP() + + @mcp.prompt(tags={"example", "test-tag"}) + def sample_prompt() -> str: + return "Hello, world!" + + prompts_dict = await mcp.get_prompts() + assert len(prompts_dict) == 1 + prompt = prompts_dict["sample_prompt"] + assert prompt.tags == {"example", "test-tag"} From 3f793d541cdba5c31f6aed6537c9a0bc6d4d6e74 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 25 Apr 2025 18:32:24 -0400 Subject: [PATCH 2/8] minor docs updates --- docs/servers/context.mdx | 2 +- docs/servers/prompts.mdx | 2 +- docs/servers/resources.mdx | 6 +++--- docs/servers/tools.mdx | 2 +- tests/server/test_server_interactions.py | 15 +++++++++++++++ 5 files changed, 21 insertions(+), 6 deletions(-) diff --git a/docs/servers/context.mdx b/docs/servers/context.mdx index e6ce5a56f..685b833b0 100644 --- a/docs/servers/context.mdx +++ b/docs/servers/context.mdx @@ -19,7 +19,7 @@ The `Context` object provides a clean interface to access MCP features within yo - **Request Information**: Access metadata about the current request - **Server Access**: When needed, access the underlying FastMCP server instance -## Accessing Context +## Accessing the Context To use the context object within your tool function, simply add a parameter to your function signature and type-hint it as `Context`. FastMCP will automatically inject the context instance when your tool is called. diff --git a/docs/servers/prompts.mdx b/docs/servers/prompts.mdx index b9a6b520c..5668846f4 100644 --- a/docs/servers/prompts.mdx +++ b/docs/servers/prompts.mdx @@ -20,7 +20,7 @@ Prompts provide parameterized message templates for LLMs. When a client requests This allows you to define consistent, reusable templates that LLMs can use across different clients and contexts. -## Defining Prompts +## Prompts ### The `@prompt` Decorator diff --git a/docs/servers/resources.mdx b/docs/servers/resources.mdx index 447e854f5..6b1fe6bd2 100644 --- a/docs/servers/resources.mdx +++ b/docs/servers/resources.mdx @@ -1,6 +1,6 @@ --- title: Resources & Templates -sidebarTitle: Resources & Templates +sidebarTitle: Resources description: Expose data sources and dynamic content generators to your MCP client. icon: database --- @@ -21,7 +21,7 @@ Resources provide read-only access to data for the LLM or client application. Wh This allows LLMs to access files, database content, configuration, or dynamically generated information relevant to the conversation. -## Defining Resources +## Resources ### The `@resource` Decorator @@ -201,7 +201,7 @@ mcp.add_resource(special_resource, key="internal://data-v2") # Will be stored a Note that this parameter is only available when using `add_resource()` directly and not through the `@resource` decorator, as URIs are provided explicitly when using the decorator. -## Defining Resource Templates +## Resource Templates Resource Templates allow clients to request resources whose content depends on parameters embedded in the URI. Define a template using the **same `@mcp.resource` decorator**, but include `{parameter_name}` placeholders in the URI string and add corresponding arguments to your function signature. diff --git a/docs/servers/tools.mdx b/docs/servers/tools.mdx index 7d4fdbf50..6eded291f 100644 --- a/docs/servers/tools.mdx +++ b/docs/servers/tools.mdx @@ -18,7 +18,7 @@ Tools in FastMCP transform regular Python functions into capabilities that LLMs This allows LLMs to perform tasks like querying databases, calling APIs, making calculations, or accessing files—extending their capabilities beyond what's in their training data. -## Defining Tools +## Tools ### The `@tool` Decorator diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py index 6ac501787..a02afd533 100644 --- a/tests/server/test_server_interactions.py +++ b/tests/server/test_server_interactions.py @@ -180,6 +180,21 @@ class TestTools: assert "title" in properties assert properties["title"]["description"] == "Optional title" + async def test_tool_with_bytes_input(self): + mcp = FastMCP() + + @mcp.tool() + def process_image(image: bytes) -> Image: + return Image(data=image) + + async with Client(mcp) as client: + result = await client.call_tool( + "process_image", {"image": b"fake png data"} + ) + assert isinstance(result[0], ImageContent) + assert result[0].mimeType == "image/png" + assert result[0].data == base64.b64encode(b"fake png data").decode() + class TestResources: async def test_text_resource(self): From 8a4aa707a3c6861af8e03adf444be823d232219e Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 25 Apr 2025 18:41:14 -0400 Subject: [PATCH 3/8] Update tools.mdx --- docs/servers/tools.mdx | 105 ++++++++++++++++++++++++++--------------- 1 file changed, 66 insertions(+), 39 deletions(-) diff --git a/docs/servers/tools.mdx b/docs/servers/tools.mdx index 6eded291f..44c2eafde 100644 --- a/docs/servers/tools.mdx +++ b/docs/servers/tools.mdx @@ -44,44 +44,31 @@ When this tool is registered, FastMCP automatically: The way you define your Python function dictates how the tool appears and behaves for the LLM client. -### Type Annotations +### Parameters -Type annotations are crucial. They: -1. Inform the LLM about the expected type for each parameter. -2. Allow FastMCP to validate the data received from the client. -3. Are used to generate the tool's input schema for the MCP protocol. +#### Annotations -FastMCP supports standard Python type annotations, including those from the `typing` module and Pydantic. +Type annotations for parameters are essential for proper tool functionality. They: +1. Inform the LLM about the expected data types for each parameter +2. Enable FastMCP to validate input data from clients +3. Generate accurate JSON schemas for the MCP protocol + +Use standard Python type annotations for parameters: ```python -from typing import Literal, Optional, Union -from pydantic import BaseModel, Field - -# Example using various type hints @mcp.tool() -def process_data( - data: list[float], # List of floats - operation: Literal["sum", "average", "max"], # Fixed choices - precision: int = 2, # Optional int with default - description: str | None = None # Optional string (can be None) +def analyze_text( + text: str, + max_tokens: int = 100, + language: str | None = None ) -> dict: - """Process numerical data with the specified operation.""" - result = 0.0 - if operation == "sum": - result = sum(data) - elif operation == "average": - result = sum(data) / len(data) if data else 0.0 - elif operation == "max": - result = float(max(data)) if data else 0.0 - - return { - "operation": operation, - "result": round(result, precision), - "description": description - } + """Analyze the provided text.""" + # Implementation... ``` -**Supported Type Annotation Examples:** +#### Supported Types + +FastMCP supports a wide range of type annotations: | Type Annotation | Example | Description | | :---------------------- | :---------------------------- | :---------------------------------- | @@ -90,30 +77,70 @@ def process_data( | Optional types | `Optional[float]`, `float\|None`| Parameters that may be null/omitted | | Union types | `str \| int`, `Union[str, int]`| Parameters accepting multiple types | | Literal types | `Literal["A", "B"]` | Parameters with specific allowed values | -| Pydantic models | `UserData` | Complex structured data (see below) | +| Pydantic models | `UserData` | Complex structured data (see Structured Inputs) | **Automatic JSON Parsing:** FastMCP intelligently handles arguments. If a client sends a string that looks like valid JSON (e.g., `"['a', 'b']"`) for a parameter hinted as a structured type (like `list[str]` or a Pydantic model), FastMCP will automatically attempt to parse the JSON string into the expected Python object before validation. This improves robustness when interacting with various clients. -### Required vs. Optional Parameters +#### Parameter Metadata -Parameters in your function signature are considered **required** unless they have a default value. +You can provide additional metadata about parameters using Pydantic's `Field` class with `Annotated`. This approach is preferred as it's more modern and keeps type hints separate from validation rules: + +```python +from typing import Annotated +from pydantic import Field + +@mcp.tool() +def process_image( + image_url: Annotated[str, Field(description="URL of the image to process")], + resize: Annotated[bool, Field(description="Whether to resize the image")] = False, + width: Annotated[int, Field(description="Target width in pixels", ge=1, le=2000)] = 800, + format: Annotated[ + Literal["jpeg", "png", "webp"], + Field(description="Output image format") + ] = "jpeg" +) -> dict: + """Process an image with optional resizing.""" + # Implementation... +``` + +You can also use the Field as a default value, though the Annotated approach is preferred: + +```python +@mcp.tool() +def search_database( + query: str = Field(description="Search query string"), + limit: int = Field(10, description="Maximum number of results", ge=1, le=100) +) -> list: + """Search the database with the provided query.""" + # Implementation... +``` + +Field provides several validation and documentation features: +- `description`: Human-readable explanation of the parameter (shown to LLMs) +- `ge`/`gt`/`le`/`lt`: Greater/less than (or equal) constraints +- `min_length`/`max_length`: String or collection length constraints +- `pattern`: Regex pattern for string validation +- `default`: Default value if parameter is omitted + +#### Optional Arguments + +FastMCP follows Python's standard function parameter conventions. Parameters without default values are required, while those with default values are optional. ```python @mcp.tool() def search_products( - query: str, # Required - no default value - max_results: int = 10, # Optional - has default value - sort_by: str = "relevance" # Optional - has default value + query: str, # Required - no default value + max_results: int = 10, # Optional - has default value + sort_by: str = "relevance", # Optional - has default value + category: str | None = None # Optional - can be None ) -> list[dict]: """Search the product catalog.""" # Implementation... - print(f"Searching for '{query}', max {max_results}, sorted by {sort_by}") - return [{"id": 1, "name": "Sample Product"}] ``` -In this example, the LLM *must* provide a `query`. If `max_results` or `sort_by` are omitted, their default values will be used. +In this example, the LLM must provide a `query` parameter, while `max_results`, `sort_by`, and `category` will use their default values if not explicitly provided. ### Structured Inputs From 5f1e1dea5b63c1836793494606a279d746978824 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 25 Apr 2025 19:05:20 -0400 Subject: [PATCH 4/8] update tool docs --- docs/servers/tools.mdx | 230 +++++++++++++++++------ tests/server/test_server_interactions.py | 14 ++ 2 files changed, 187 insertions(+), 57 deletions(-) diff --git a/docs/servers/tools.mdx b/docs/servers/tools.mdx index 44c2eafde..888593f33 100644 --- a/docs/servers/tools.mdx +++ b/docs/servers/tools.mdx @@ -66,23 +66,6 @@ def analyze_text( # Implementation... ``` -#### Supported Types - -FastMCP supports a wide range of type annotations: - -| Type Annotation | Example | Description | -| :---------------------- | :---------------------------- | :---------------------------------- | -| Basic types | `int`, `float`, `str`, `bool` | Simple scalar values | -| Container types | `list[str]`, `dict[str, int]` | Collections of items | -| Optional types | `Optional[float]`, `float\|None`| Parameters that may be null/omitted | -| Union types | `str \| int`, `Union[str, int]`| Parameters accepting multiple types | -| Literal types | `Literal["A", "B"]` | Parameters with specific allowed values | -| Pydantic models | `UserData` | Complex structured data (see Structured Inputs) | - - -**Automatic JSON Parsing:** FastMCP intelligently handles arguments. If a client sends a string that looks like valid JSON (e.g., `"['a', 'b']"`) for a parameter hinted as a structured type (like `list[str]` or a Pydantic model), FastMCP will automatically attempt to parse the JSON string into the expected Python object before validation. This improves robustness when interacting with various clients. - - #### Parameter Metadata You can provide additional metadata about parameters using Pydantic's `Field` class with `Annotated`. This approach is preferred as it's more modern and keeps type hints separate from validation rules: @@ -124,6 +107,20 @@ Field provides several validation and documentation features: - `pattern`: Regex pattern for string validation - `default`: Default value if parameter is omitted +#### Supported Types + +FastMCP supports a wide range of type annotations: + +| Type Annotation | Example | Description | +| :---------------------- | :---------------------------- | :---------------------------------- | +| Basic types | `int`, `float`, `str`, `bool` | Simple scalar values - see [Built-in Types](#built-in-types) | +| Binary data | `bytes` | Binary content - see [Binary Data Handling](#binary-data-handling) | +| Collection types | `list[str]`, `dict[str, int]`, `set[int]` | Collections of items - see [Collection Types](#collection-types) | +| Optional types | `float \| None`, `Optional[float]`| Parameters that may be null/omitted - see [Union and Optional Types](#union-and-optional-types) | +| Union types | `str \| int`, `Union[str, int]`| Parameters accepting multiple types - see [Union and Optional Types](#union-and-optional-types) | +| Literal types | `Literal["A", "B"]` | Parameters with specific allowed values - see [Literal Types](#literal-types) | +| Pydantic models | `UserData` | Complex structured data - see [Pydantic Models](#pydantic-models) | + #### Optional Arguments FastMCP follows Python's standard function parameter conventions. Parameters without default values are required, while those with default values are optional. @@ -142,44 +139,6 @@ def search_products( In this example, the LLM must provide a `query` parameter, while `max_results`, `sort_by`, and `category` will use their default values if not explicitly provided. -### Structured Inputs - -For tools requiring complex, nested, or well-validated inputs, use Pydantic models. Define a `BaseModel` and use it as a type hint for a parameter. - -```python -from pydantic import BaseModel, Field -from typing import Optional -from datetime import date - -class ReservationRequest(BaseModel): - guest_name: str = Field(description="Full name of the guest making the reservation.") - check_in: date - check_out: date - room_type: Literal["standard", "deluxe", "suite"] = Field(default="standard", description="Type of room requested.") - guests: int = Field(gt=0, description="Number of guests (must be positive).") - special_requests: Optional[str] = Field(default=None, description="Any special requests for the stay.") - -@mcp.tool() -def make_reservation(request: ReservationRequest) -> dict: - """Creates a new hotel reservation based on the provided details.""" - # Pydantic automatically validates the incoming 'request' data - # against the ReservationRequest model before this function runs. - print(f"Making reservation for {request.guest_name}...") - # Implementation... - return { - "reservation_id": "R12345", - "status": "confirmed", - "guest": request.guest_name, - "dates": f"{request.check_in} to {request.check_out}" - } -``` - -Using Pydantic models provides: -- Clear, self-documenting structure for complex inputs. -- Built-in data validation (e.g., `gt=0`, date parsing). -- Automatic generation of detailed JSON schemas for the LLM. -- Easy handling of optional fields and default values. - ### Metadata While FastMCP infers the name and description from your function, you can override these and add tags using arguments to the `@mcp.tool` decorator: @@ -293,7 +252,7 @@ FastMCP automatically catches exceptions raised within your tool function: Using informative exceptions helps the LLM understand failures and react appropriately. -### Using Context in Tools +### Accessing MCP Context Tools can access MCP features like logging, reading resources, or reporting progress through the `Context` object. To use it, add a parameter to your tool function with the type hint `Context`. @@ -365,4 +324,161 @@ The duplicate behavior options are: - `"warn"` (default): Logs a warning and the new tool replaces the old one. - `"error"`: Raises a `ValueError`, preventing the duplicate registration. - `"replace"`: Silently replaces the existing tool with the new one. -- `"ignore"`: Keeps the original tool and ignores the new registration attempt. \ No newline at end of file +- `"ignore"`: Keeps the original tool and ignores the new registration attempt. + +## Parameter Types + +FastMCP supports a wide variety of parameter types to give you flexibility when designing your tools. When clients send parameters, FastMCP will attempt to coerce values into the appropriate type when possible (for example, parsing JSON strings into structured types). + +### Built-in Types + +The most common parameter types are Python's built-in scalar types: + +```python +@mcp.tool() +def process_values( + name: str, # Text data + count: int, # Integer numbers + amount: float, # Floating point numbers + enabled: bool # Boolean values (True/False) +): + """Process various value types.""" + # Implementation... +``` + +These types provide clear expectations to the LLM about what values are acceptable and allow FastMCP to validate inputs properly. Even if a client provides a string like "42", it will be coerced to an integer for parameters annotated as `int`. + +### Collection Types + +For structured data collections, FastMCP supports standard Python collection types: + +```python +@mcp.tool() +def analyze_data( + values: list[float], # List of numbers + labels: list[str], # List of strings + properties: dict[str, str], # Dictionary with string keys and values + mixed_data: dict[str, list[int]] # Nested collections +): + """Analyze collections of data.""" + # Implementation... +``` + +Collection types can be nested and combined to represent complex data structures. If a client sends a JSON string like `"[1.5, 2.5, 3.5]"` for a `list[float]` parameter, FastMCP will automatically parse and convert it. + +### Union and Optional Types + +For parameters that can accept multiple types or may be omitted: + +```python +@mcp.tool() +def flexible_search( + query: str | int, # Can be either string or integer + filters: dict[str, str] | None = None, # Optional dictionary + sort_field: str | None = None # Optional string +): + """Search with flexible parameter types.""" + # Implementation... +``` + +Modern Python syntax (`str | int`) is preferred over older `Union[str, int]` forms. Similarly, `str | None` is preferred over `Optional[str]`. + +### Literal Types + +When a parameter must be one of a predefined set of values: + +```python +from typing import Literal + +@mcp.tool() +def sort_data( + data: list[float], + order: Literal["ascending", "descending"] = "ascending", + algorithm: Literal["quicksort", "mergesort", "heapsort"] = "quicksort" +): + """Sort data using specified order and algorithm.""" + # Implementation... +``` + +Literal types help LLMs understand exactly which values are acceptable and provide validation for incoming parameters. + +### Binary Data Handling + +There are two approaches to handling binary data in tool parameters: + +#### Using bytes type + +```python +@mcp.tool() +def process_binary(data: bytes): + """Process binary data directly. + + The client can send a binary string, which will be + converted directly to bytes. + """ + # Implementation using binary data + data_length = len(data) + # ... +``` + +When you annotate a parameter as `bytes`, FastMCP will: +- Convert raw strings directly to bytes +- Validate that the input can be properly represented as bytes + +FastMCP does not automatically decode base64-encoded strings for bytes parameters. If you need to accept base64-encoded data, you should handle the decoding manually as shown below. + +#### Using base64-encoded strings + +```python +from typing import Annotated +from pydantic import Field + +@mcp.tool() +def process_image_data( + image_data: Annotated[str, Field(description="Base64-encoded image data")] +): + """Process an image from base64-encoded string. + + The client is expected to provide base64-encoded data as a string. + You'll need to decode it manually. + """ + # Manual base64 decoding + import base64 + binary_data = base64.b64decode(image_data) + # Process binary_data... +``` + +This approach is recommended when you expect to receive base64-encoded binary data from clients. + +### Pydantic Models + +For complex, structured data with nested fields and validation, use Pydantic models: + +```python +from pydantic import BaseModel, Field +from typing import Optional + +class User(BaseModel): + username: str + email: str = Field(description="User's email address") + age: int | None = None + is_active: bool = True + +@mcp.tool() +def create_user(user: User): + """Create a new user in the system.""" + # The input is automatically validated against the User model + # Even if provided as a JSON string or dict + # Implementation... +``` + +Using Pydantic models provides: +- Clear, self-documenting structure for complex inputs +- Built-in data validation +- Automatic generation of detailed JSON schemas for the LLM +- Automatic conversion from dict/JSON input + +Clients can provide data for Pydantic model parameters as either: +- A JSON object (string) +- A dictionary with the appropriate structure +- Nested parameters in the appropriate format \ No newline at end of file diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py index a02afd533..ac7fffde6 100644 --- a/tests/server/test_server_interactions.py +++ b/tests/server/test_server_interactions.py @@ -195,6 +195,20 @@ class TestTools: assert result[0].mimeType == "image/png" assert result[0].data == base64.b64encode(b"fake png data").decode() + async def test_tool_with_invalid_input(self): + mcp = FastMCP() + + @mcp.tool() + def my_tool(x: int) -> int: + return x + 1 + + async with Client(mcp) as client: + with pytest.raises( + ClientError, + match="Input should be a valid integer, unable to parse string as an integer", + ): + await client.call_tool("my_tool", {"x": "not an int"}) + class TestResources: async def test_text_resource(self): From cd97b1c20133db64424e815f5a6b92ac2edcf66d Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 25 Apr 2025 19:30:43 -0400 Subject: [PATCH 5/8] Document and test input types --- docs/servers/tools.mdx | 141 +++++++++++-- tests/server/test_server.py | 33 +++ tests/server/test_server_interactions.py | 254 ++++++++++++++++++++++- 3 files changed, 414 insertions(+), 14 deletions(-) diff --git a/docs/servers/tools.mdx b/docs/servers/tools.mdx index 888593f33..2d75b9688 100644 --- a/docs/servers/tools.mdx +++ b/docs/servers/tools.mdx @@ -118,9 +118,11 @@ FastMCP supports a wide range of type annotations: | Collection types | `list[str]`, `dict[str, int]`, `set[int]` | Collections of items - see [Collection Types](#collection-types) | | Optional types | `float \| None`, `Optional[float]`| Parameters that may be null/omitted - see [Union and Optional Types](#union-and-optional-types) | | Union types | `str \| int`, `Union[str, int]`| Parameters accepting multiple types - see [Union and Optional Types](#union-and-optional-types) | -| Literal types | `Literal["A", "B"]` | Parameters with specific allowed values - see [Literal Types](#literal-types) | +| Constrained types | `Literal["A", "B"]`, `Enum` | Parameters with specific allowed values - see [Constrained Types](#constrained-types) | | Pydantic models | `UserData` | Complex structured data - see [Pydantic Models](#pydantic-models) | +For additional type annotations not listed here, see the [Parameter Types](#parameter-types) section below for more detailed information and examples. + #### Optional Arguments FastMCP follows Python's standard function parameter conventions. Parameters without default values are required, while those with default values are optional. @@ -198,6 +200,8 @@ FastMCP automatically converts the value returned by your function into the appr - **`fastmcp.Image`**: A helper class for easily returning image data. Sent as `ImageContent`. - **`None`**: Results in an empty response (no content is sent back to the client). +FastMCP will attempt to serialize other types to a string if possible. + ```python from fastmcp import FastMCP, Image import io @@ -328,7 +332,12 @@ The duplicate behavior options are: ## Parameter Types -FastMCP supports a wide variety of parameter types to give you flexibility when designing your tools. When clients send parameters, FastMCP will attempt to coerce values into the appropriate type when possible (for example, parsing JSON strings into structured types). +FastMCP supports a wide variety of parameter types to give you flexibility when designing your tools. + + + +FastMCP supports **type coercion** when possible. This means that if a client sends data that doesn't match the expected type, FastMCP will attempt to convert it to the appropriate type. For example, if a client sends a string for a parameter annotated as `int`, FastMCP will attempt to convert it to an integer. If the conversion is not possible, FastMCP will return a validation error. + ### Built-in Types @@ -350,21 +359,28 @@ These types provide clear expectations to the LLM about what values are acceptab ### Collection Types -For structured data collections, FastMCP supports standard Python collection types: +FastMCP supports all standard Python collection types: ```python @mcp.tool() def analyze_data( values: list[float], # List of numbers - labels: list[str], # List of strings properties: dict[str, str], # Dictionary with string keys and values + unique_ids: set[int], # Set of unique integers + coordinates: tuple[float, float], # Tuple with fixed structure mixed_data: dict[str, list[int]] # Nested collections ): """Analyze collections of data.""" # Implementation... ``` -Collection types can be nested and combined to represent complex data structures. If a client sends a JSON string like `"[1.5, 2.5, 3.5]"` for a `list[float]` parameter, FastMCP will automatically parse and convert it. +All collection types can be used as parameter annotations: +- `list[T]` - Ordered sequence of items +- `dict[K, V]` - Key-value mapping +- `set[T]` - Unordered collection of unique items +- `tuple[T1, T2, ...]` - Fixed-length sequence with potentially different types + +Collection types can be nested and combined to represent complex data structures. JSON strings that match the expected structure will be automatically parsed and converted to the appropriate Python collection type. ### Union and Optional Types @@ -383,9 +399,13 @@ def flexible_search( Modern Python syntax (`str | int`) is preferred over older `Union[str, int]` forms. Similarly, `str | None` is preferred over `Optional[str]`. -### Literal Types +### Constrained Types -When a parameter must be one of a predefined set of values: +When a parameter must be one of a predefined set of values, you can use either Literal types or Enums: + +#### Literals + +Literals constrain parameters to a specific set of values: ```python from typing import Literal @@ -396,17 +416,49 @@ def sort_data( order: Literal["ascending", "descending"] = "ascending", algorithm: Literal["quicksort", "mergesort", "heapsort"] = "quicksort" ): - """Sort data using specified order and algorithm.""" + """Sort data using specific options.""" # Implementation... ``` -Literal types help LLMs understand exactly which values are acceptable and provide validation for incoming parameters. +Literal types: +- Specify exact allowable values directly in the type annotation +- Help LLMs understand exactly which values are acceptable +- Provide input validation (errors for invalid values) +- Create clear schemas for clients -### Binary Data Handling +#### Enums + +For more structured sets of constrained values, use Python's Enum class: + +```python +from enum import Enum + +class Color(Enum): + RED = "red" + GREEN = "green" + BLUE = "blue" + +@mcp.tool() +def process_image( + image_path: str, + color_filter: Color = Color.RED +): + """Process an image with a color filter.""" + # Implementation... + # color_filter will be a Color enum member +``` + +When using Enum types: +- Clients should provide the enum's value (e.g., "red"), not the enum member name (e.g., "RED") +- FastMCP automatically coerces the string value into the appropriate Enum object +- Your function receives the actual Enum member (e.g., `Color.RED`) +- Validation errors are raised for values not in the enum + +### Binary Data There are two approaches to handling binary data in tool parameters: -#### Using bytes type +#### Bytes ```python @mcp.tool() @@ -427,7 +479,7 @@ When you annotate a parameter as `bytes`, FastMCP will: FastMCP does not automatically decode base64-encoded strings for bytes parameters. If you need to accept base64-encoded data, you should handle the decoding manually as shown below. -#### Using base64-encoded strings +#### Base64-encoded strings ```python from typing import Annotated @@ -481,4 +533,67 @@ Using Pydantic models provides: Clients can provide data for Pydantic model parameters as either: - A JSON object (string) - A dictionary with the appropriate structure -- Nested parameters in the appropriate format \ No newline at end of file +- Nested parameters in the appropriate format + +### Pydantic Fields + +FastMCP supports robust parameter validation through Pydantic's `Field` class. This is especially useful to ensure that input values meet specific requirements beyond just their type. + +Note that fields can be used *outside* Pydantic models to provide metadata and validation constraints. The preferred approach is using `Annotated` with `Field`: + +```python +from typing import Annotated +from pydantic import Field + +@mcp.tool() +def analyze_metrics( + # Numbers with range constraints + count: Annotated[int, Field(ge=0, le=100)], # 0 <= count <= 100 + ratio: Annotated[float, Field(gt=0, lt=1.0)], # 0 < ratio < 1.0 + + # String with pattern and length constraints + user_id: Annotated[str, Field( + pattern=r"^[A-Z]{2}\d{4}$", # Must match regex pattern + description="User ID in format XX0000" + )], + + # String with length constraints + comment: Annotated[str, Field(min_length=3, max_length=500)] = "", + + # Numeric constraints + factor: Annotated[int, Field(multiple_of=5)] = 10, # Must be multiple of 5 +): + """Analyze metrics with validated parameters.""" + # Implementation... +``` + +You can also use `Field` as a default value, though the `Annotated` approach is preferred: + +```python +@mcp.tool() +def validate_data( + # Value constraints + age: int = Field(ge=0, lt=120), # 0 <= age < 120 + + # String constraints + email: str = Field(pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$"), # Email pattern + + # Collection constraints + tags: list[str] = Field(min_length=1, max_length=10) # 1-10 tags +): + """Process data with field validations.""" + # Implementation... +``` + +Common validation options include: + +| Validation | Type | Description | +| :--------- | :--- | :---------- | +| `ge`, `gt` | Number | Greater than (or equal) constraint | +| `le`, `lt` | Number | Less than (or equal) constraint | +| `multiple_of` | Number | Value must be a multiple of this number | +| `min_length`, `max_length` | String, List, etc. | Length constraints | +| `pattern` | String | Regular expression pattern constraint | +| `description` | Any | Human-readable description (appears in schema) | + +When a client sends invalid data, FastMCP will return a validation error explaining why the parameter failed validation. diff --git a/tests/server/test_server.py b/tests/server/test_server.py index 7c4782eba..6a80c03ab 100644 --- a/tests/server/test_server.py +++ b/tests/server/test_server.py @@ -1,8 +1,11 @@ +from typing import Annotated + import pytest from mcp.types import ( TextContent, TextResourceContents, ) +from pydantic import Field from fastmcp import Client, FastMCP from fastmcp.exceptions import ClientError, NotFoundError @@ -239,6 +242,36 @@ class TestToolDecorator: # Original name should not be registered assert "multiply" not in tools + async def test_tool_with_annotated_arguments(self): + """Test that tools with annotated arguments work correctly.""" + mcp = FastMCP() + + @mcp.tool() + def add( + x: Annotated[int, Field(description="x is an int")], + y: Annotated[str, Field(description="y is not an int")], + ) -> None: + pass + + tool = (await mcp.get_tools())["add"] + assert tool.parameters["properties"]["x"]["description"] == "x is an int" + assert tool.parameters["properties"]["y"]["description"] == "y is not an int" + + async def test_tool_with_field_defaults(self): + """Test that tools with annotated arguments work correctly.""" + mcp = FastMCP() + + @mcp.tool() + def add( + x: int = Field(description="x is an int"), + y: str = Field(description="y is not an int"), + ) -> None: + pass + + tool = (await mcp.get_tools())["add"] + assert tool.parameters["properties"]["x"]["description"] == "x is an int" + assert tool.parameters["properties"]["y"]["description"] == "y is not an int" + class TestResourceDecorator: async def test_no_resources_before_decorator(self): diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py index ac7fffde6..cd56a0b09 100644 --- a/tests/server/test_server_interactions.py +++ b/tests/server/test_server_interactions.py @@ -1,6 +1,8 @@ import base64 import json +from enum import Enum from pathlib import Path +from typing import Annotated, Literal import pytest from mcp.types import ( @@ -157,7 +159,32 @@ class TestTools: assert isinstance(content3, TextContent) assert content3.text == "direct content" - async def test_parameter_descriptions(self): + async def test_parameter_descriptions_with_field_annotations(self): + mcp = FastMCP("Test Server") + + @mcp.tool() + def greet( + name: Annotated[str, Field(description="The name to greet")], + title: Annotated[str, Field(description="Optional title", default="")], + ) -> str: + """A greeting tool""" + return f"Hello {title} {name}" + + async with Client(mcp) as client: + tools = await client.list_tools() + assert len(tools) == 1 + tool = tools[0] + + # Check that parameter descriptions are present in the schema + properties = tool.inputSchema["properties"] + assert "name" in properties + assert properties["name"]["description"] == "The name to greet" + assert "title" in properties + assert properties["title"]["description"] == "Optional title" + assert properties["title"]["default"] == "" + assert tool.inputSchema["required"] == ["name"] + + async def test_parameter_descriptions_with_field_defaults(self): mcp = FastMCP("Test Server") @mcp.tool() @@ -179,6 +206,8 @@ class TestTools: assert properties["name"]["description"] == "The name to greet" assert "title" in properties assert properties["title"]["description"] == "Optional title" + assert properties["title"]["default"] == "" + assert tool.inputSchema["required"] == ["name"] async def test_tool_with_bytes_input(self): mcp = FastMCP() @@ -209,6 +238,229 @@ class TestTools: ): await client.call_tool("my_tool", {"x": "not an int"}) + async def test_tool_int_coercion(self): + """Test string-to-int type coercion.""" + mcp = FastMCP() + + @mcp.tool() + def add_one(x: int) -> int: + return x + 1 + + async with Client(mcp) as client: + # String with integer value should be coerced to int + result = await client.call_tool("add_one", {"x": "42"}) + assert isinstance(result[0], TextContent) + assert result[0].text == "43" + + async def test_tool_bool_coercion(self): + """Test string-to-bool type coercion.""" + mcp = FastMCP() + + @mcp.tool() + def toggle(flag: bool) -> bool: + return not flag + + async with Client(mcp) as client: + # String with boolean value should be coerced to bool + result = await client.call_tool("toggle", {"flag": "true"}) + assert isinstance(result[0], TextContent) + assert result[0].text == "false" + + result = await client.call_tool("toggle", {"flag": "false"}) + assert isinstance(result[0], TextContent) + assert result[0].text == "true" + + async def test_tool_list_coercion(self): + """Test JSON string to collection type coercion.""" + mcp = FastMCP() + + @mcp.tool() + def process_list(items: list[int]) -> int: + return sum(items) + + async with Client(mcp) as client: + # JSON array string should be coerced to list + result = await client.call_tool( + "process_list", {"items": "[1, 2, 3, 4, 5]"} + ) + assert isinstance(result[0], TextContent) + assert result[0].text == "15" + + async def test_tool_list_coercion_error(self): + """Test that a list coercion error is raised if the input is not a valid list.""" + mcp = FastMCP() + + @mcp.tool() + def process_list(items: list[int]) -> int: + return sum(items) + + async with Client(mcp) as client: + with pytest.raises( + ClientError, + match="Input should be a valid list", + ): + await client.call_tool("process_list", {"items": "['a', 'b', 3]"}) + + async def test_tool_dict_coercion(self): + """Test JSON string to dict type coercion.""" + mcp = FastMCP() + + @mcp.tool() + def process_dict(data: dict[str, int]) -> int: + return sum(data.values()) + + async with Client(mcp) as client: + # JSON object string should be coerced to dict + result = await client.call_tool( + "process_dict", {"data": '{"a": 1, "b": "2", "c": 3}'} + ) + assert isinstance(result[0], TextContent) + assert result[0].text == "6" + + async def test_tool_set_coercion(self): + """Test JSON string to set type coercion.""" + mcp = FastMCP() + + @mcp.tool() + def process_set(items: set[int]) -> int: + assert isinstance(items, set) + return sum(items) + + async with Client(mcp) as client: + result = await client.call_tool("process_set", {"items": "[1, 2, 3, 4, 5]"}) + assert isinstance(result[0], TextContent) + assert result[0].text == "15" + + async def test_tool_tuple_coercion(self): + """Test JSON string to tuple type coercion.""" + mcp = FastMCP() + + @mcp.tool() + def process_tuple(items: tuple[int, str]) -> int: + assert isinstance(items, tuple) + return items[0] + len(items[1]) + + async with Client(mcp) as client: + result = await client.call_tool("process_tuple", {"items": '["1", "two"]'}) + assert isinstance(result[0], TextContent) + assert result[0].text == "4" + + async def test_annotated_field_validation(self): + mcp = FastMCP() + + @mcp.tool() + def analyze(x: Annotated[int, Field(ge=1)]) -> None: + pass + + async with Client(mcp) as client: + with pytest.raises( + ClientError, + match="Input should be greater than or equal to 1", + ): + await client.call_tool("analyze", {"x": 0}) + + async def test_default_field_validation(self): + mcp = FastMCP() + + @mcp.tool() + def analyze(x: int = Field(ge=1)) -> None: + pass + + async with Client(mcp) as client: + with pytest.raises( + ClientError, + match="Input should be greater than or equal to 1", + ): + await client.call_tool("analyze", {"x": 0}) + + async def test_default_field_is_still_required_if_no_default_specified(self): + mcp = FastMCP() + + @mcp.tool() + def analyze(x: int = Field()) -> None: + pass + + async with Client(mcp) as client: + with pytest.raises(ClientError, match="Field required"): + await client.call_tool("analyze", {}) + + async def test_literal_type_validation_error(self): + mcp = FastMCP() + + @mcp.tool() + def analyze(x: Literal["a", "b"]) -> None: + pass + + async with Client(mcp) as client: + with pytest.raises(ClientError, match="Input should be 'a' or 'b'"): + await client.call_tool("analyze", {"x": "c"}) + + async def test_literal_type_validation_success(self): + mcp = FastMCP() + + @mcp.tool() + def analyze(x: Literal["a", "b"]) -> str: + return x + + async with Client(mcp) as client: + result = await client.call_tool("analyze", {"x": "a"}) + assert isinstance(result[0], TextContent) + assert result[0].text == "a" + + async def test_enum_type_validation_error(self): + mcp = FastMCP() + + class MyEnum(Enum): + RED = "red" + GREEN = "green" + BLUE = "blue" + + @mcp.tool() + def analyze(x: MyEnum) -> str: + return x.value + + async with Client(mcp) as client: + with pytest.raises( + ClientError, match="Input should be 'red', 'green' or 'blue'" + ): + await client.call_tool("analyze", {"x": "some-color"}) + + async def test_enum_type_validation_success(self): + mcp = FastMCP() + + class MyEnum(Enum): + RED = "red" + GREEN = "green" + BLUE = "blue" + + @mcp.tool() + def analyze(x: MyEnum) -> str: + return x.value + + async with Client(mcp) as client: + result = await client.call_tool("analyze", {"x": "red"}) + assert isinstance(result[0], TextContent) + assert result[0].text == "red" + + async def test_union_type_validation(self): + mcp = FastMCP() + + @mcp.tool() + def analyze(x: int | float) -> str: + return str(x) + + async with Client(mcp) as client: + result = await client.call_tool("analyze", {"x": 1}) + assert isinstance(result[0], TextContent) + assert result[0].text == "1" + + result = await client.call_tool("analyze", {"x": 1.0}) + assert isinstance(result[0], TextContent) + assert result[0].text == "1.0" + + with pytest.raises(ClientError, match="2 validation errors for analyze"): + await client.call_tool("analyze", {"x": "not a number"}) + class TestResources: async def test_text_resource(self): From 882a10a5e3947ba797ef7a4f387c8066f01f21f6 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 25 Apr 2025 19:48:23 -0400 Subject: [PATCH 6/8] Update docs and tests --- docs/servers/tools.mdx | 70 ++++++++++- tests/server/test_server_interactions.py | 143 +++++++++++++++++++++++ 2 files changed, 209 insertions(+), 4 deletions(-) diff --git a/docs/servers/tools.mdx b/docs/servers/tools.mdx index 2d75b9688..71c24dfbe 100644 --- a/docs/servers/tools.mdx +++ b/docs/servers/tools.mdx @@ -109,16 +109,19 @@ Field provides several validation and documentation features: #### Supported Types -FastMCP supports a wide range of type annotations: +FastMCP supports a wide range of type annotations, including all Pydantic types: | Type Annotation | Example | Description | | :---------------------- | :---------------------------- | :---------------------------------- | | Basic types | `int`, `float`, `str`, `bool` | Simple scalar values - see [Built-in Types](#built-in-types) | -| Binary data | `bytes` | Binary content - see [Binary Data Handling](#binary-data-handling) | +| Binary data | `bytes` | Binary content - see [Binary Data](#binary-data) | +| Date and Time | `datetime`, `date`, `timedelta` | Date and time objects - see [Date and Time Types](#date-and-time-types) | | Collection types | `list[str]`, `dict[str, int]`, `set[int]` | Collections of items - see [Collection Types](#collection-types) | | Optional types | `float \| None`, `Optional[float]`| Parameters that may be null/omitted - see [Union and Optional Types](#union-and-optional-types) | | Union types | `str \| int`, `Union[str, int]`| Parameters accepting multiple types - see [Union and Optional Types](#union-and-optional-types) | | Constrained types | `Literal["A", "B"]`, `Enum` | Parameters with specific allowed values - see [Constrained Types](#constrained-types) | +| Paths | `Path` | File system paths - see [Paths](#paths) | +| UUIDs | `UUID` | Universally unique identifiers - see [UUIDs](#uuids) | | Pydantic models | `UserData` | Complex structured data - see [Pydantic Models](#pydantic-models) | For additional type annotations not listed here, see the [Parameter Types](#parameter-types) section below for more detailed information and examples. @@ -334,11 +337,10 @@ The duplicate behavior options are: FastMCP supports a wide variety of parameter types to give you flexibility when designing your tools. - +FastMCP generally supports all types that Pydantic supports as fields, including all Pydantic custom types. This means you can use any type that can be validated and parsed by Pydantic in your tool parameters. FastMCP supports **type coercion** when possible. This means that if a client sends data that doesn't match the expected type, FastMCP will attempt to convert it to the appropriate type. For example, if a client sends a string for a parameter annotated as `int`, FastMCP will attempt to convert it to an integer. If the conversion is not possible, FastMCP will return a validation error. - ### Built-in Types The most common parameter types are Python's built-in scalar types: @@ -357,6 +359,32 @@ def process_values( These types provide clear expectations to the LLM about what values are acceptable and allow FastMCP to validate inputs properly. Even if a client provides a string like "42", it will be coerced to an integer for parameters annotated as `int`. +### Date and Time Types + +FastMCP supports various date and time types from the `datetime` module: + +```python +from datetime import datetime, date, timedelta + +@mcp.tool() +def process_date_time( + event_date: date, # ISO format date string or date object + event_time: datetime, # ISO format datetime string or datetime object + duration: timedelta = timedelta(hours=1) # Integer seconds or timedelta +) -> str: + """Process date and time information.""" + # Types are automatically converted from strings + assert isinstance(event_date, date) + assert isinstance(event_time, datetime) + assert isinstance(duration, timedelta) + + return f"Event on {event_date} at {event_time} for {duration}" +``` + +- `datetime` - Accepts ISO format strings (e.g., "2023-04-15T14:30:00") +- `date` - Accepts ISO format date strings (e.g., "2023-04-15") +- `timedelta` - Accepts integer seconds or timedelta objects + ### Collection Types FastMCP supports all standard Python collection types: @@ -502,6 +530,40 @@ def process_image_data( This approach is recommended when you expect to receive base64-encoded binary data from clients. +### Paths + +The `Path` type from the `pathlib` module can be used for file system paths: + +```python +from pathlib import Path + +@mcp.tool() +def process_file(path: Path) -> str: + """Process a file at the given path.""" + assert isinstance(path, Path) # Path is properly converted + return f"Processing file at {path}" +``` + +When a client sends a string path, FastMCP automatically converts it to a `Path` object. + +### UUIDs + +The `UUID` type from the `uuid` module can be used for unique identifiers: + +```python +import uuid + +@mcp.tool() +def process_item( + item_id: uuid.UUID # String UUID or UUID object +) -> str: + """Process an item with the given UUID.""" + assert isinstance(item_id, uuid.UUID) # Properly converted to UUID + return f"Processing item {item_id}" +``` + +When a client sends a string UUID (e.g., "123e4567-e89b-12d3-a456-426614174000"), FastMCP automatically converts it to a `UUID` object. + ### Pydantic Models For complex, structured data with nested fields and validation, use Pydantic models: diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py index cd56a0b09..3144f710b 100644 --- a/tests/server/test_server_interactions.py +++ b/tests/server/test_server_interactions.py @@ -1,5 +1,7 @@ import base64 +import datetime import json +import uuid from enum import Enum from pathlib import Path from typing import Annotated, Literal @@ -159,6 +161,8 @@ class TestTools: assert isinstance(content3, TextContent) assert content3.text == "direct content" + +class TestToolParameters: async def test_parameter_descriptions_with_field_annotations(self): mcp = FastMCP("Test Server") @@ -461,6 +465,145 @@ class TestTools: with pytest.raises(ClientError, match="2 validation errors for analyze"): await client.call_tool("analyze", {"x": "not a number"}) + async def test_path_type(self): + mcp = FastMCP() + + @mcp.tool() + def send_path(path: Path) -> str: + assert isinstance(path, Path) + return str(path) + + async with Client(mcp) as client: + result = await client.call_tool("send_path", {"path": "/tmp/test.txt"}) + assert isinstance(result[0], TextContent) + assert result[0].text == "/tmp/test.txt" + + async def test_path_type_error(self): + mcp = FastMCP() + + @mcp.tool() + def send_path(path: Path) -> str: + return str(path) + + async with Client(mcp) as client: + with pytest.raises(ClientError, match="Input is not a valid path"): + await client.call_tool("send_path", {"path": 1}) + + async def test_uuid_type(self): + mcp = FastMCP() + + @mcp.tool() + def send_uuid(x: uuid.UUID) -> str: + assert isinstance(x, uuid.UUID) + return str(x) + + test_uuid = uuid.uuid4() + + async with Client(mcp) as client: + result = await client.call_tool("send_uuid", {"x": test_uuid}) + assert isinstance(result[0], TextContent) + assert result[0].text == str(test_uuid) + + async def test_uuid_type_error(self): + mcp = FastMCP() + + @mcp.tool() + def send_uuid(x: uuid.UUID) -> str: + return str(x) + + async with Client(mcp) as client: + with pytest.raises(ClientError, match="Input should be a valid UUID"): + await client.call_tool("send_uuid", {"x": "not a uuid"}) + + async def test_datetime_type(self): + mcp = FastMCP() + + @mcp.tool() + def send_datetime(x: datetime.datetime) -> str: + return x.isoformat() + + async with Client(mcp) as client: + result = await client.call_tool( + "send_datetime", {"x": datetime.datetime.now()} + ) + assert isinstance(result[0], TextContent) + assert result[0].text == datetime.datetime.now().isoformat() + + async def test_datetime_type_parse_string(self): + mcp = FastMCP() + + @mcp.tool() + def send_datetime(x: datetime.datetime) -> str: + return x.isoformat() + + async with Client(mcp) as client: + result = await client.call_tool( + "send_datetime", {"x": "2021-01-01T00:00:00"} + ) + assert isinstance(result[0], TextContent) + assert result[0].text == "2021-01-01T00:00:00" + + async def test_datetime_type_error(self): + mcp = FastMCP() + + @mcp.tool() + def send_datetime(x: datetime.datetime) -> str: + return x.isoformat() + + async with Client(mcp) as client: + with pytest.raises(ClientError, match="Input should be a valid datetime"): + await client.call_tool("send_datetime", {"x": "not a datetime"}) + + async def test_date_type(self): + mcp = FastMCP() + + @mcp.tool() + def send_date(x: datetime.date) -> str: + return x.isoformat() + + async with Client(mcp) as client: + result = await client.call_tool("send_date", {"x": datetime.date.today()}) + assert isinstance(result[0], TextContent) + assert result[0].text == datetime.date.today().isoformat() + + async def test_date_type_parse_string(self): + mcp = FastMCP() + + @mcp.tool() + def send_date(x: datetime.date) -> str: + return x.isoformat() + + async with Client(mcp) as client: + result = await client.call_tool("send_date", {"x": "2021-01-01"}) + assert isinstance(result[0], TextContent) + assert result[0].text == "2021-01-01" + + async def test_timedelta_type(self): + mcp = FastMCP() + + @mcp.tool() + def send_timedelta(x: datetime.timedelta) -> str: + return str(x) + + async with Client(mcp) as client: + result = await client.call_tool( + "send_timedelta", {"x": datetime.timedelta(days=1)} + ) + assert isinstance(result[0], TextContent) + assert result[0].text == "1 day, 0:00:00" + + async def test_timedelta_type_parse_int(self): + mcp = FastMCP() + + @mcp.tool() + def send_timedelta(x: datetime.timedelta) -> str: + return str(x) + + async with Client(mcp) as client: + result = await client.call_tool("send_timedelta", {"x": 1000}) + assert isinstance(result[0], TextContent) + assert result[0].text == "0:16:40" + class TestResources: async def test_text_resource(self): From 8205e1eefb647c865296a8888fd31dd0ee8c911c Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 25 Apr 2025 19:51:16 -0400 Subject: [PATCH 7/8] Improve datetime test to avoid now --- tests/server/test_server_interactions.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py index 3144f710b..9f588a475 100644 --- a/tests/server/test_server_interactions.py +++ b/tests/server/test_server_interactions.py @@ -522,12 +522,12 @@ class TestToolParameters: def send_datetime(x: datetime.datetime) -> str: return x.isoformat() + dt = datetime.datetime(2025, 4, 25, 1, 2, 3) + async with Client(mcp) as client: - result = await client.call_tool( - "send_datetime", {"x": datetime.datetime.now()} - ) + result = await client.call_tool("send_datetime", {"x": dt}) assert isinstance(result[0], TextContent) - assert result[0].text == datetime.datetime.now().isoformat() + assert result[0].text == dt.isoformat() async def test_datetime_type_parse_string(self): mcp = FastMCP() From 702bc68a4c026736306026e2c41cb037171f45c3 Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 25 Apr 2025 20:05:04 -0400 Subject: [PATCH 8/8] platform independent path --- tests/server/test_server_interactions.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py index 9f588a475..94f7ebdd0 100644 --- a/tests/server/test_server_interactions.py +++ b/tests/server/test_server_interactions.py @@ -473,10 +473,13 @@ class TestToolParameters: assert isinstance(path, Path) return str(path) + # Use a platform-independent path + test_path = Path("tmp") / "test.txt" + async with Client(mcp) as client: - result = await client.call_tool("send_path", {"path": "/tmp/test.txt"}) + result = await client.call_tool("send_path", {"path": str(test_path)}) assert isinstance(result[0], TextContent) - assert result[0].text == "/tmp/test.txt" + assert result[0].text == str(test_path) async def test_path_type_error(self): mcp = FastMCP()