ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-23 14:04:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #261 from jlowin/server-tests

Browse source 
Significant documentation and test expansion for tool input types
This commit is contained in:
Jeremiah Lowin 2025-04-25 20:06:55 -04:00 • committed by GitHub
commit 8348ed6224
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1721 additions and 926 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/servers/context.mdx
View file

@ -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.

2
docs/servers/prompts.mdx
View file

@ -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

6
docs/servers/resources.mdx
View file

@ -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.

496
docs/servers/tools.mdx
View file

@ -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
@ -44,114 +44,105 @@ 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:**
#### 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:
```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
#### Supported Types
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 |
| 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 below) |
| Basic types | `int`, `float`, `str`, `bool` | Simple scalar values - see [Built-in Types](#built-in-types) |
| 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) |
<Tip>
**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.
</Tip>
For additional type annotations not listed here, see the [Parameter Types](#parameter-types) section below for more detailed information and examples.
### Required vs. Optional Parameters
#### Optional Arguments
Parameters in your function signature are considered **required** unless they have a default value.
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.
### 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.
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.
### Metadata
@ -212,6 +203,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
@ -266,7 +259,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`.
@ -338,4 +331,331 @@ 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.
- `"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.
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:
```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`.
### 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:
```python
@mcp.tool()
def analyze_data(
values: list[float], # List of numbers
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...
```
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
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]`.
### Constrained Types
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
@mcp.tool()
def sort_data(
data: list[float],
order: Literal["ascending", "descending"] = "ascending",
algorithm: Literal["quicksort", "mergesort", "heapsort"] = "quicksort"
):
"""Sort data using specific options."""
# Implementation...
```
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
#### 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:
#### Bytes
```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.
#### 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.
### 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:
```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
### 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.

871
tests/server/test_server.py
View file

@ -1,25 +1,14 @@
import base64
import json
from pathlib import Path
from typing import TYPE_CHECKING
from typing import Annotated
import pytest
from mcp.types import (
BlobResourceContents,
ImageContent,
TextContent,
TextResourceContents,
)
from pydantic import AnyUrl, Field
from pydantic import 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:
@ -253,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):
@ -715,825 +734,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"}

1278
tests/server/test_server_interactions.py Normal file
View file

File diff suppressed because it is too large Load diff