Add tests

This commit is contained in:
Jeremiah Lowin 2025-06-14 09:22:29 -04:00
commit 06dcff3b3b
5 changed files with 362 additions and 28 deletions

View file

@ -2,6 +2,7 @@
<!-- omit in toc -->
# FastMCP v2 🚀
<strong>The fast, Pythonic way to build MCP servers and clients.</strong>
*FastMCP is made with 💙 by [Prefect](https://www.prefect.io/)*
@ -15,10 +16,11 @@
</div>
> [!Note]
>
> #### Beyond the Protocol
>
>
> FastMCP is the standard framework for working with the Model Context Protocol. FastMCP 1.0 was incorporated into the [official MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk) in 2024.
>
>
> This is FastMCP 2.0, the **actively maintained version** that provides a complete toolkit for working with the MCP ecosystem.
>
> FastMCP 2.0 has a comprehensive set of features that go far beyond the core MCP specification, all in service of providing **the simplest path to production**. These include deployment, auth, clients, server proxying and composition, generating servers from REST APIs, dynamic tool rewriting, built-in testing tools, integrations, and more.
@ -45,6 +47,7 @@ if __name__ == "__main__":
```
Run the server locally:
```bash
fastmcp run server.py
```
@ -53,9 +56,10 @@ fastmcp run server.py
FastMCP's complete documentation is available at **[gofastmcp.com](https://gofastmcp.com)**, including detailed guides, API references, and advanced patterns. This readme provides only a high-level overview.
Documentation is also available in [llms.txt format](https://llmstxt.org/), which is a simple markdown standard that LLMs can consume easily.
Documentation is also available in [llms.txt format](https://llmstxt.org/), which is a simple markdown standard that LLMs can consume easily.
There are two ways to access the LLM-friendly documentation:
- [`llms.txt`](https://gofastmcp.com/llms.txt) is essentially a sitemap, listing all the pages in the documentation.
- [`llms-full.txt`](https://gofastmcp.com/llms-full.txt) contains the entire documentation. Note this may exceed the context window of your LLM.
@ -145,7 +149,7 @@ Learn more in the [**FastMCP Server Documentation**](https://gofastmcp.com/serve
### Tools
Tools allow LLMs to perform actions by executing your Python functions (sync or async). Ideal for computations, API calls, or side effects (like `POST`/`PUT`). FastMCP handles schema generation from type hints and docstrings. Tools can return various types, including text, JSON-serializable objects, and even images using the [`fastmcp.Image`](https://gofastmcp.com/servers/tools#return-values) helper.
Tools allow LLMs to perform actions by executing your Python functions (sync or async). Ideal for computations, API calls, or side effects (like `POST`/`PUT`). FastMCP handles schema generation from type hints and docstrings. Tools can return various types, including text, JSON-serializable objects, and even images or audio aided by the FastMCP media helper classes.
```python
@mcp.tool
@ -191,12 +195,13 @@ Learn more in the [**Prompts Documentation**](https://gofastmcp.com/servers/prom
### Context
Access MCP session capabilities within your tools, resources, or prompts by adding a `ctx: Context` parameter. Context provides methods for:
* **Logging:** Log messages to MCP clients with `ctx.info()`, `ctx.error()`, etc.
* **LLM Sampling:** Use `ctx.sample()` to request completions from the client's LLM.
* **HTTP Request:** Use `ctx.http_request()` to make HTTP requests to other servers.
* **Resource Access:** Use `ctx.read_resource()` to access resources on the server
* **Progress Reporting:** Use `ctx.report_progress()` to report progress to the client.
* and more...
- **Logging:** Log messages to MCP clients with `ctx.info()`, `ctx.error()`, etc.
- **LLM Sampling:** Use `ctx.sample()` to request completions from the client's LLM.
- **HTTP Request:** Use `ctx.http_request()` to make HTTP requests to other servers.
- **Resource Access:** Use `ctx.read_resource()` to access resources on the server
- **Progress Reporting:** Use `ctx.report_progress()` to report progress to the client.
- and more...
To access the context, add a parameter annotated as `Context` to any mcp-decorated function. FastMCP will automatically inject the correct context object when the function is called.
@ -336,16 +341,19 @@ if __name__ == "__main__":
FastMCP supports three transport protocols:
**STDIO (Default)**: Best for local tools and command-line scripts.
```python
mcp.run(transport="stdio") # Default, so transport argument is optional
```
**Streamable HTTP**: Recommended for web deployments.
```python
mcp.run(transport="streamable-http", host="127.0.0.1", port=8000, path="/mcp")
```
**SSE**: For compatibility with existing SSE clients.
```python
mcp.run(transport="sse", host="127.0.0.1", port=8000)
```
@ -358,22 +366,26 @@ Contributions are the core of open source! We welcome improvements and features.
### Prerequisites
* Python 3.10+
* [uv](https://docs.astral.sh/uv/) (Recommended for environment management)
- Python 3.10+
- [uv](https://docs.astral.sh/uv/) (Recommended for environment management)
### Setup
1. Clone the repository:
1. Clone the repository:
```bash
git clone https://github.com/jlowin/fastmcp.git
cd fastmcp
```
2. Create and sync the environment:
2. Create and sync the environment:
```bash
uv sync
```
This installs all dependencies, including dev tools.
3. Activate the virtual environment (e.g., `source .venv/bin/activate` or via your IDE).
### Unit Tests
@ -381,10 +393,13 @@ Contributions are the core of open source! We welcome improvements and features.
FastMCP has a comprehensive unit test suite. All PRs must introduce or update tests as appropriate and pass the full suite.
Run tests using pytest:
```bash
pytest
```
or if you want an overview of the code coverage
```bash
uv run pytest --cov=src --cov=examples --cov-report=html
```
@ -394,10 +409,13 @@ uv run pytest --cov=src --cov=examples --cov-report=html
FastMCP uses `pre-commit` for code formatting, linting, and type-checking. All PRs must pass these checks (they run automatically in CI).
Install the hooks locally:
```bash
uv run pre-commit install
```
The hooks will now run automatically on `git commit`. You can also run them manually at any time:
```bash
pre-commit run --all-files
# or via uv
@ -406,11 +424,11 @@ uv run pre-commit run --all-files
### Pull Requests
1. Fork the repository on GitHub.
2. Create a feature branch from `main`.
3. Make your changes, including tests and documentation updates.
4. Ensure tests and pre-commit hooks pass.
5. Commit your changes and push to your fork.
6. Open a pull request against the `main` branch of `jlowin/fastmcp`.
1. Fork the repository on GitHub.
2. Create a feature branch from `main`.
3. Make your changes, including tests and documentation updates.
4. Ensure tests and pre-commit hooks pass.
5. Commit your changes and push to your fork.
6. Open a pull request against the `main` branch of `jlowin/fastmcp`.
Please open an issue or discussion for questions or suggestions before starting significant work!
Please open an issue or discussion for questions or suggestions before starting significant work!

View file

@ -256,7 +256,9 @@ FastMCP automatically converts the value returned by your function into the appr
- **`str`**: Sent as `TextContent`.
- **`dict`, `list`, Pydantic `BaseModel`**: Serialized to a JSON string and sent as `TextContent`.
- **`bytes`**: Base64 encoded and sent as `BlobResourceContents` (often within an `EmbeddedResource`).
- **`fastmcp.Image`**: A helper class for easily returning image data. Sent as `ImageContent`.
- **`fastmcp.utilities.types.Image`**: A helper class for easily returning image data. Sent as `ImageContent`.
- **`fastmcp.utilities.types.Audio`**: A helper class for easily returning audio data. Sent as `AudioContent`.
- **A list of any of the above**: Automatically converts each item appropriately.
- **`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.

View file

@ -10,6 +10,7 @@ import pydantic_core
import pytest
from mcp import McpError
from mcp.types import (
AudioContent,
EmbeddedResource,
ImageContent,
TextContent,
@ -24,7 +25,7 @@ from fastmcp.prompts.prompt import Prompt, PromptMessage
from fastmcp.resources import FileResource, ResourceTemplate
from fastmcp.resources.resource import FunctionResource
from fastmcp.tools.tool import Tool
from fastmcp.utilities.types import Image
from fastmcp.utilities.types import Audio, Image
@pytest.fixture
@ -47,6 +48,10 @@ def tool_server():
def image_tool(path: str) -> Image:
return Image(path)
@mcp.tool
def audio_tool(path: str) -> Audio:
return Audio(path)
@mcp.tool
def mixed_content_tool() -> list[TextContent | ImageContent]:
return [
@ -63,6 +68,15 @@ def tool_server():
TextContent(type="text", text="direct content"),
]
@mcp.tool
def mixed_audio_list_fn(audio_path: str) -> list:
return [
"text message",
Audio(audio_path),
{"key": "value"},
TextContent(type="text", text="direct content"),
]
return mcp
@ -74,7 +88,7 @@ class TestTools:
async def test_list_tools(self, tool_server: FastMCP):
async with Client(tool_server) as client:
assert len(await client.list_tools()) == 6
assert len(await client.list_tools()) == 8
async def test_call_tool(self, tool_server: FastMCP):
async with Client(tool_server) as client:
@ -115,6 +129,104 @@ class TestTools:
result = await client.call_tool("list_tool", {})
assert result[0].text == '[\n "x",\n 2\n]' # type: ignore[attr-defined]
async def test_image(self, tmp_path: Path):
mcp = FastMCP()
@mcp.tool
def image_tool(path: str) -> Image:
return Image(path)
# Create a test image
image_path = tmp_path / "test.png"
image_path.write_bytes(b"fake png data")
async with Client(mcp) 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_audio(self, tmp_path: Path):
mcp = FastMCP()
@mcp.tool
def audio_tool(path: str) -> Audio:
return Audio(path)
# Create a test audio file
audio_path = tmp_path / "test.wav"
audio_path.write_bytes(b"fake wav data")
async with Client(mcp) as client:
result = await client.call_tool("audio_tool", {"path": str(audio_path)})
content = result[0]
assert isinstance(content, AudioContent)
assert content.type == "audio"
assert content.mimeType == "audio/wav"
# Verify base64 encoding
decoded = base64.b64decode(content.data)
assert decoded == b"fake wav data"
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_tool_mixed_list_with_audio(
self, tool_server: FastMCP, tmp_path: Path
):
"""Test that lists containing Audio objects and other types are handled
correctly. Note that the non-MCP content will be grouped together."""
# Create a test audio file
audio_path = tmp_path / "test.wav"
audio_path.write_bytes(b"test audio data")
async with Client(tool_server) as client:
result = await client.call_tool(
"mixed_audio_list_fn", {"audio_path": str(audio_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 audio conversion
content2 = result[1]
assert isinstance(content2, AudioContent)
assert content2.mimeType == "audio/wav"
assert base64.b64decode(content2.data) == b"test audio data"
# Check direct TextContent
content3 = result[2]
assert isinstance(content3, TextContent)
assert content3.text == "direct content"
class TestToolTags:
def create_server(self, include_tags=None, exclude_tags=None):
@ -269,6 +381,27 @@ class TestToolReturnTypes:
decoded = base64.b64decode(content.data)
assert decoded == b"fake png data"
async def test_audio(self, tmp_path: Path):
mcp = FastMCP()
@mcp.tool
def audio_tool(path: str) -> Audio:
return Audio(path)
# Create a test audio file
audio_path = tmp_path / "test.wav"
audio_path.write_bytes(b"fake wav data")
async with Client(mcp) as client:
result = await client.call_tool("audio_tool", {"path": str(audio_path)})
content = result[0]
assert isinstance(content, AudioContent)
assert content.type == "audio"
assert content.mimeType == "audio/wav"
# Verify base64 encoding
decoded = base64.b64decode(content.data)
assert decoded == b"fake wav 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", {})
@ -309,6 +442,34 @@ class TestToolReturnTypes:
assert isinstance(content3, TextContent)
assert content3.text == "direct content"
async def test_tool_mixed_list_with_audio(
self, tool_server: FastMCP, tmp_path: Path
):
"""Test that lists containing Audio objects and other types are handled
correctly. Note that the non-MCP content will be grouped together."""
# Create a test audio file
audio_path = tmp_path / "test.wav"
audio_path.write_bytes(b"test audio data")
async with Client(tool_server) as client:
result = await client.call_tool(
"mixed_audio_list_fn", {"audio_path": str(audio_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 audio conversion
content2 = result[1]
assert isinstance(content2, AudioContent)
assert content2.mimeType == "audio/wav"
assert base64.b64decode(content2.data) == b"test audio data"
# Check direct TextContent
content3 = result[2]
assert isinstance(content3, TextContent)
assert content3.text == "direct content"
class TestToolParameters:
async def test_parameter_descriptions_with_field_annotations(self):
@ -972,7 +1133,7 @@ class TestResourceTags:
assert {r.name for r in resources} == set()
async def test_exclude_tags_some_resources(self):
mcp = self.create_server(exclude_tags={"a", "z"})
mcp = self.create_server(exclude_tags={"a"})
async with Client(mcp) as client:
resources = await client.list_resources()
@ -1403,6 +1564,9 @@ class TestResourceTemplatesTags:
with pytest.raises(McpError, match="Unknown resource"):
await client.read_resource("resource://1/x")
result = await client.read_resource("resource://2/x")
assert result[0].text == "Template resource 2: x" # type: ignore[attr-defined]
class TestResourceTemplateContext:
async def test_resource_template_context(self):
@ -1496,6 +1660,9 @@ class TestResourceTemplateEnabled:
templates = await client.list_resource_templates()
assert len(templates) == 0
with pytest.raises(McpError, match="Unknown resource"):
await client.read_resource(AnyUrl("resource://test"))
async def test_get_template_and_disable(self):
mcp = FastMCP()

View file

@ -1,5 +1,11 @@
import pytest
from mcp.types import EmbeddedResource, ImageContent, TextContent, TextResourceContents
from mcp.types import (
AudioContent,
EmbeddedResource,
ImageContent,
TextContent,
TextResourceContents,
)
from pydantic import AnyUrl, BaseModel
from fastmcp import FastMCP
@ -7,7 +13,7 @@ from fastmcp.client import Client
from fastmcp.exceptions import ToolError
from fastmcp.tools.tool import Tool, _convert_to_content
from fastmcp.utilities.tests import temporary_settings
from fastmcp.utilities.types import Image
from fastmcp.utilities.types import Audio, Image
class TestToolFromFunction:
@ -98,6 +104,16 @@ class TestToolFromFunction:
assert tool.parameters["properties"]["data"]["type"] == "string"
assert isinstance(result[0], ImageContent)
async def test_tool_with_audio_return(self):
def audio_tool(data: bytes) -> Audio:
return Audio(data=data)
tool = Tool.from_function(audio_tool)
result = await tool.run({"data": "test.wav"})
assert tool.parameters["properties"]["data"]["type"] == "string"
assert isinstance(result[0], AudioContent)
def test_non_callable_fn(self):
with pytest.raises(TypeError, match="not a callable object"):
Tool.from_function(1) # type: ignore
@ -441,6 +457,17 @@ class TestConvertResultToContent:
assert isinstance(result[0], ImageContent)
assert result[0].data == "ZmFrZWltYWdlZGF0YQ=="
def test_audio_object_result(self):
"""Test that an Audio object is converted to AudioContent."""
audio_obj = Audio(data=b"fakeaudiodata")
result = _convert_to_content(audio_obj)
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], AudioContent)
assert result[0].data == "ZmFrZWF1ZGlvZGF0YQ=="
def test_basic_type_result(self):
"""Test that a basic type is converted to TextContent."""
result = _convert_to_content(123)
@ -525,6 +552,28 @@ class TestConvertResultToContent:
image_item = next(item for item in result if isinstance(item, ImageContent))
assert image_item.data == "ZmFrZWltYWdlZGF0YQ=="
def test_list_of_mixed_types_with_audio(self):
"""Test that a list of mixed types including Audio is converted correctly."""
content1 = TextContent(type="text", text="hello")
audio_obj = Audio(data=b"fakeaudiodata")
basic_data = {"a": 1}
result = _convert_to_content([content1, audio_obj, basic_data])
assert isinstance(result, list)
assert len(result) == 3
text_content_count = sum(isinstance(item, TextContent) for item in result)
audio_content_count = sum(isinstance(item, AudioContent) for item in result)
assert text_content_count == 2
assert audio_content_count == 1
text_item = next(item for item in result if isinstance(item, TextContent))
assert text_item.text == '{\n "a": 1\n}'
audio_item = next(item for item in result if isinstance(item, AudioContent))
assert audio_item.data == "ZmFrZWF1ZGlvZGF0YQ=="
def test_empty_list(self):
"""Test that an empty list results in an empty list."""
result = _convert_to_content([])

View file

@ -5,6 +5,7 @@ from typing import Annotated, Any
import pytest
from fastmcp.utilities.types import (
Audio,
Image,
find_kwarg_by_type,
is_class_member_of_type,
@ -202,6 +203,103 @@ class TestImage:
img.to_image_content()
class TestAudio:
def test_audio_initialization_with_path(self):
"""Test audio initialization with a path."""
# Mock test - we're not actually going to read a file
audio = Audio(path="test.wav")
assert audio.path is not None
assert audio.data is None
assert audio._mime_type == "audio/wav"
def test_audio_initialization_with_data(self):
"""Test audio initialization with data."""
audio = Audio(data=b"test")
assert audio.path is None
assert audio.data == b"test"
assert audio._mime_type == "audio/wav" # Default for raw data
def test_audio_initialization_with_format(self):
"""Test audio initialization with a specific format."""
audio = Audio(data=b"test", format="mp3")
assert audio._mime_type == "audio/mp3"
def test_missing_data_and_path_raises_error(self):
"""Test that error is raised when neither path nor data is provided."""
with pytest.raises(ValueError, match="Either path or data must be provided"):
Audio()
def test_both_data_and_path_raises_error(self):
"""Test that error is raised when both path and data are provided."""
with pytest.raises(
ValueError, match="Only one of path or data can be provided"
):
Audio(path="test.wav", data=b"test")
def test_get_mime_type_from_path(self, tmp_path):
"""Test MIME type detection from file extension."""
extensions = {
".wav": "audio/wav",
".mp3": "audio/mpeg",
".ogg": "audio/ogg",
".m4a": "audio/mp4",
".flac": "audio/flac",
".unknown": "application/octet-stream",
}
for ext, mime in extensions.items():
path = tmp_path / f"test{ext}"
path.write_bytes(b"fake audio data")
audio = Audio(path=path)
assert audio._mime_type == mime
def test_to_audio_content(self, tmp_path, monkeypatch):
"""Test conversion to AudioContent."""
# Test with path
audio_path = tmp_path / "test.wav"
test_data = b"fake audio data"
audio_path.write_bytes(test_data)
audio = Audio(path=audio_path)
content = audio.to_audio_content()
assert content.type == "audio"
assert content.mimeType == "audio/wav"
assert content.data == base64.b64encode(test_data).decode()
# Test with data
audio = Audio(data=test_data, format="mp3")
content = audio.to_audio_content()
assert content.type == "audio"
assert content.mimeType == "audio/mp3"
assert content.data == base64.b64encode(test_data).decode()
def test_to_audio_content_error(self, monkeypatch):
"""Test error case in to_audio_content."""
# Create an Audio with neither path nor data (shouldn't happen due to __init__ checks,
# but testing the method's own error handling)
audio = Audio(data=b"test")
monkeypatch.setattr(audio, "path", None)
monkeypatch.setattr(audio, "data", None)
with pytest.raises(ValueError, match="No audio data available"):
audio.to_audio_content()
def test_to_audio_content_with_override_mime_type(self, tmp_path):
"""Test conversion to AudioContent with override MIME type."""
audio_path = tmp_path / "test.wav"
test_data = b"fake audio data"
audio_path.write_bytes(test_data)
audio = Audio(path=audio_path)
content = audio.to_audio_content(mime_type="audio/custom")
assert content.type == "audio"
assert content.mimeType == "audio/custom"
assert content.data == base64.b64encode(test_data).decode()
class TestFindKwargByType:
def test_exact_type_match(self):
"""Test finding parameter with exact type match."""