ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Use upstream FastMCP

Browse source
This commit is contained in:
Jeremiah Lowin 2025-04-01 18:35:57 -04:00
commit ca810729d7
38 changed files with 166 additions and 5138 deletions
Download patch file Download diff file Expand all files Collapse all files

88
examples/fastapi_converter.py Normal file
View file

@ -0,0 +1,88 @@
"""Example demonstrating FastAPI to FastMCP conversion."""
from fastapi import FastAPI, Body, Path
from pydantic import BaseModel
from typing import List, Optional
from fastmcp.convert.fastapi import fastapi_to_fastmcp
# Create a sample FastAPI app
app = FastAPI(title="Todo API")
class TodoItem(BaseModel):
"""A simple todo item."""
id: Optional[int] = None
title: str
description: str
completed: bool = False
# In-memory database
todos = {}
todo_id_counter = 1
@app.get("/todos", response_model=List[TodoItem])
async def get_todos():
"""Retrieve all todo items."""
return list(todos.values())
@app.get("/todos/{todo_id}", response_model=TodoItem)
async def get_todo(todo_id: int = Path(..., description="The ID of the todo item")):
"""Retrieve a specific todo item by ID."""
if todo_id not in todos:
return {"error": f"Todo {todo_id} not found"}
return todos[todo_id]
@app.post("/todos", response_model=TodoItem)
async def create_todo(
todo: TodoItem = Body(..., description="The todo item to create"),
):
"""Create a new todo item."""
global todo_id_counter
todo.id = todo_id_counter
todos[todo_id_counter] = todo
todo_id_counter += 1
return todo
@app.put("/todos/{todo_id}", response_model=TodoItem)
async def update_todo(
todo_id: int = Path(..., description="The ID of the todo item"),
todo: TodoItem = Body(..., description="The updated todo item"),
):
"""Update an existing todo item."""
if todo_id not in todos:
return {"error": f"Todo {todo_id} not found"}
todo.id = todo_id
todos[todo_id] = todo
return todo
@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int = Path(..., description="The ID of the todo item")):
"""Delete a todo item."""
if todo_id not in todos:
return {"error": f"Todo {todo_id} not found"}
del todos[todo_id]
return {"message": f"Todo {todo_id} deleted"}
# Convert the FastAPI app to a FastMCP server
mcp = fastapi_to_fastmcp(app)
if __name__ == "__main__":
# Run the FastMCP server
mcp.run()
# Alternatively, you can run the original FastAPI app with uvicorn:
# import uvicorn
# uvicorn.run(app, host="0.0.0.0", port=8000)

19
pyproject.toml
View file

@ -1,16 +1,9 @@
[project]
name = "fastmcp"
dynamic = ["version"]
description = "A more ergonomic interface for MCP servers"
description = "An ergonomic interface for MCP servers"
authors = [{ name = "Jeremiah Lowin" }]
dependencies = [
"httpx>=0.26.0",
"mcp>=1.0.0,<2.0.0",
"pydantic-settings>=2.6.1",
"pydantic>=2.5.3,<3.0.0",
"typer>=0.9.0",
"python-dotenv>=1.0.1",
]
dependencies = ["mcp>=1.6.0,<2.0.0", "rich>=13.9.4"]
requires-python = ">=3.10"
readme = "README.md"
license = { text = "MIT" }
@ -22,8 +15,8 @@ fastmcp = "fastmcp.cli:app"
requires = ["hatchling>=1.21.0", "hatch-vcs>=0.4.0"]
build-backend = "hatchling.build"
[project.optional-dependencies]
tests = [
[dependency-groups]
dev = [
"pre-commit",
"pyright>=1.1.389",
"pytest>=8.3.3",
@ -31,8 +24,10 @@ tests = [
"pytest-flakefinder",
"pytest-xdist>=3.6.1",
"ruff",
"copychat>=0.5.2",
"ipython>=8.12.3",
"pdbpp>=0.10.3",
]
dev = ["fastmcp[tests]", "copychat>=0.5.2", "ipython>=8.12.3", "pdbpp>=0.10.3"]
[tool.pytest.ini_options]
asyncio_mode = "auto"

3
src/fastmcp/__init__.py
View file

@ -1,8 +1,7 @@
"""FastMCP - A more ergonomic interface for MCP servers."""
from importlib.metadata import version
from .server import FastMCP, Context
from .utilities.types import Image
from mcp.server.fastmcp import FastMCP, Context, Image
__version__ = version("fastmcp")
__all__ = ["FastMCP", "Context", "Image"]

7
src/fastmcp/cli/__init__.py
View file

@ -1,7 +0,0 @@
"""FastMCP CLI package."""
from .cli import app
if __name__ == "__main__":
app()

138
src/fastmcp/cli/claude.py
View file

@ -1,138 +0,0 @@
"""Claude app integration utilities."""
import json
import sys
from pathlib import Path
from typing import Optional, Dict
from ..utilities.logging import get_logger
logger = get_logger(__name__)
def get_claude_config_path() -> Path | None:
"""Get the Claude config directory based on platform."""
if sys.platform == "win32":
path = Path(Path.home(), "AppData", "Roaming", "Claude")
elif sys.platform == "darwin":
path = Path(Path.home(), "Library", "Application Support", "Claude")
else:
return None
if path.exists():
return path
return None
def update_claude_config(
file_spec: str,
server_name: str,
*,
with_editable: Optional[Path] = None,
with_packages: Optional[list[str]] = None,
env_vars: Optional[Dict[str, str]] = None,
) -> bool:
"""Add or update a FastMCP server in Claude's configuration.
Args:
file_spec: Path to the server file, optionally with :object suffix
server_name: Name for the server in Claude's config
with_editable: Optional directory to install in editable mode
with_packages: Optional list of additional packages to install
env_vars: Optional dictionary of environment variables. These are merged with
any existing variables, with new values taking precedence.
Raises:
RuntimeError: If Claude Desktop's config directory is not found, indicating
Claude Desktop may not be installed or properly set up.
"""
config_dir = get_claude_config_path()
if not config_dir:
raise RuntimeError(
"Claude Desktop config directory not found. Please ensure Claude Desktop "
"is installed and has been run at least once to initialize its configuration."
)
config_file = config_dir / "claude_desktop_config.json"
if not config_file.exists():
try:
config_file.write_text("{}")
except Exception as e:
logger.error(
"Failed to create Claude config file",
extra={
"error": str(e),
"config_file": str(config_file),
},
)
return False
try:
config = json.loads(config_file.read_text())
if "mcpServers" not in config:
config["mcpServers"] = {}
# Always preserve existing env vars and merge with new ones
if (
server_name in config["mcpServers"]
and "env" in config["mcpServers"][server_name]
):
existing_env = config["mcpServers"][server_name]["env"]
if env_vars:
# New vars take precedence over existing ones
env_vars = {**existing_env, **env_vars}
else:
env_vars = existing_env
# Build uv run command
args = ["run"]
# Collect all packages in a set to deduplicate
packages = {"fastmcp"}
if with_packages:
packages.update(pkg for pkg in with_packages if pkg)
# Add all packages with --with
for pkg in sorted(packages):
args.extend(["--with", pkg])
if with_editable:
args.extend(["--with-editable", str(with_editable)])
# Convert file path to absolute before adding to command
# Split off any :object suffix first
if ":" in file_spec:
file_path, server_object = file_spec.rsplit(":", 1)
file_spec = f"{Path(file_path).resolve()}:{server_object}"
else:
file_spec = str(Path(file_spec).resolve())
# Add fastmcp run command
args.extend(["fastmcp", "run", file_spec])
server_config = {
"command": "uv",
"args": args,
}
# Add environment variables if specified
if env_vars:
server_config["env"] = env_vars
config["mcpServers"][server_name] = server_config
config_file.write_text(json.dumps(config, indent=2))
logger.info(
f"Added server '{server_name}' to Claude config",
extra={"config_file": str(config_file)},
)
return True
except Exception as e:
logger.error(
"Failed to update Claude config",
extra={
"error": str(e),
"config_file": str(config_file),
},
)
return False

452
src/fastmcp/cli/cli.py
View file

@ -1,452 +0,0 @@
"""FastMCP CLI tools."""
import importlib.metadata
import importlib.util
import os
import subprocess
import sys
from pathlib import Path
from typing import Dict, Optional, Tuple
import dotenv
import typer
from typing_extensions import Annotated
from fastmcp.cli import claude
from fastmcp.utilities.logging import get_logger
logger = get_logger("cli")
app = typer.Typer(
name="fastmcp",
help="FastMCP development tools",
add_completion=False,
no_args_is_help=True, # Show help if no args provided
)
def _get_npx_command():
"""Get the correct npx command for the current platform."""
if sys.platform == "win32":
# Try both npx.cmd and npx.exe on Windows
for cmd in ["npx.cmd", "npx.exe", "npx"]:
try:
subprocess.run(
[cmd, "--version"], check=True, capture_output=True, shell=True
)
return cmd
except subprocess.CalledProcessError:
continue
return None
return "npx" # On Unix-like systems, just use npx
def _parse_env_var(env_var: str) -> Tuple[str, str]:
"""Parse environment variable string in format KEY=VALUE."""
if "=" not in env_var:
logger.error(
f"Invalid environment variable format: {env_var}. Must be KEY=VALUE"
)
sys.exit(1)
key, value = env_var.split("=", 1)
return key.strip(), value.strip()
def _build_uv_command(
file_spec: str,
with_editable: Optional[Path] = None,
with_packages: Optional[list[str]] = None,
) -> list[str]:
"""Build the uv run command that runs a FastMCP server through fastmcp run."""
cmd = ["uv"]
cmd.extend(["run", "--with", "fastmcp"])
if with_editable:
cmd.extend(["--with-editable", str(with_editable)])
if with_packages:
for pkg in with_packages:
if pkg:
cmd.extend(["--with", pkg])
# Add fastmcp run command
cmd.extend(["fastmcp", "run", file_spec])
return cmd
def _parse_file_path(file_spec: str) -> Tuple[Path, Optional[str]]:
"""Parse a file path that may include a server object specification.
Args:
file_spec: Path to file, optionally with :object suffix
Returns:
Tuple of (file_path, server_object)
"""
# First check if we have a Windows path (e.g., C:\...)
has_windows_drive = len(file_spec) > 1 and file_spec[1] == ":"
# Split on the last colon, but only if it's not part of the Windows drive letter
# and there's actually another colon in the string after the drive letter
if ":" in (file_spec[2:] if has_windows_drive else file_spec):
file_str, server_object = file_spec.rsplit(":", 1)
else:
file_str, server_object = file_spec, None
# Resolve the file path
file_path = Path(file_str).expanduser().resolve()
if not file_path.exists():
logger.error(f"File not found: {file_path}")
sys.exit(1)
if not file_path.is_file():
logger.error(f"Not a file: {file_path}")
sys.exit(1)
return file_path, server_object
def _import_server(file: Path, server_object: Optional[str] = None):
"""Import a FastMCP server from a file.
Args:
file: Path to the file
server_object: Optional object name in format "module:object" or just "object"
Returns:
The server object
"""
# Add parent directory to Python path so imports can be resolved
file_dir = str(file.parent)
if file_dir not in sys.path:
sys.path.insert(0, file_dir)
# Import the module
spec = importlib.util.spec_from_file_location("server_module", file)
if not spec or not spec.loader:
logger.error("Could not load module", extra={"file": str(file)})
sys.exit(1)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# If no object specified, try common server names
if not server_object:
# Look for the most common server object names
for name in ["mcp", "server", "app"]:
if hasattr(module, name):
return getattr(module, name)
logger.error(
f"No server object found in {file}. Please either:\n"
"1. Use a standard variable name (mcp, server, or app)\n"
"2. Specify the object name with file:object syntax",
extra={"file": str(file)},
)
sys.exit(1)
# Handle module:object syntax
if ":" in server_object:
module_name, object_name = server_object.split(":", 1)
try:
server_module = importlib.import_module(module_name)
server = getattr(server_module, object_name, None)
except ImportError:
logger.error(
f"Could not import module '{module_name}'",
extra={"file": str(file)},
)
sys.exit(1)
else:
# Just object name
server = getattr(module, server_object, None)
if server is None:
logger.error(
f"Server object '{server_object}' not found",
extra={"file": str(file)},
)
sys.exit(1)
return server
@app.command()
def version() -> None:
"""Show the FastMCP version."""
try:
version = importlib.metadata.version("fastmcp")
print(f"FastMCP version {version}")
except importlib.metadata.PackageNotFoundError:
print("FastMCP version unknown (package not installed)")
sys.exit(1)
@app.command()
def dev(
file_spec: str = typer.Argument(
...,
help="Python file to run, optionally with :object suffix",
),
with_editable: Annotated[
Optional[Path],
typer.Option(
"--with-editable",
"-e",
help="Directory containing pyproject.toml to install in editable mode",
exists=True,
file_okay=False,
resolve_path=True,
),
] = None,
with_packages: Annotated[
list[str],
typer.Option(
"--with",
help="Additional packages to install",
),
] = [],
) -> None:
"""Run a FastMCP server with the MCP Inspector."""
file, server_object = _parse_file_path(file_spec)
logger.debug(
"Starting dev server",
extra={
"file": str(file),
"server_object": server_object,
"with_editable": str(with_editable) if with_editable else None,
"with_packages": with_packages,
},
)
try:
# Import server to get dependencies
server = _import_server(file, server_object)
if hasattr(server, "dependencies"):
with_packages = list(set(with_packages + server.dependencies))
uv_cmd = _build_uv_command(file_spec, with_editable, with_packages)
# Get the correct npx command
npx_cmd = _get_npx_command()
if not npx_cmd:
logger.error(
"npx not found. Please ensure Node.js and npm are properly installed "
"and added to your system PATH."
)
sys.exit(1)
# Run the MCP Inspector command with shell=True on Windows
shell = sys.platform == "win32"
process = subprocess.run(
[npx_cmd, "@modelcontextprotocol/inspector"] + uv_cmd,
check=True,
shell=shell,
env=dict(os.environ.items()), # Convert to list of tuples for env update
)
sys.exit(process.returncode)
except subprocess.CalledProcessError as e:
logger.error(
"Dev server failed",
extra={
"file": str(file),
"error": str(e),
"returncode": e.returncode,
},
)
sys.exit(e.returncode)
except FileNotFoundError:
logger.error(
"npx not found. Please ensure Node.js and npm are properly installed "
"and added to your system PATH. You may need to restart your terminal "
"after installation.",
extra={"file": str(file)},
)
sys.exit(1)
@app.command()
def run(
file_spec: str = typer.Argument(
...,
help="Python file to run, optionally with :object suffix",
),
transport: Annotated[
Optional[str],
typer.Option(
"--transport",
"-t",
help="Transport protocol to use (stdio or sse)",
),
] = None,
) -> None:
"""Run a FastMCP server.
The server can be specified in two ways:
1. Module approach: server.py - runs the module directly, expecting a server.run() call
2. Import approach: server.py:app - imports and runs the specified server object
Note: This command runs the server directly. You are responsible for ensuring
all dependencies are available. For dependency management, use fastmcp install
or fastmcp dev instead.
"""
file, server_object = _parse_file_path(file_spec)
logger.debug(
"Running server",
extra={
"file": str(file),
"server_object": server_object,
"transport": transport,
},
)
try:
# Import and get server object
server = _import_server(file, server_object)
# Run the server
kwargs = {}
if transport:
kwargs["transport"] = transport
server.run(**kwargs)
except Exception as e:
logger.error(
f"Failed to run server: {e}",
extra={
"file": str(file),
"error": str(e),
},
)
sys.exit(1)
@app.command()
def install(
file_spec: str = typer.Argument(
...,
help="Python file to run, optionally with :object suffix",
),
server_name: Annotated[
Optional[str],
typer.Option(
"--name",
"-n",
help="Custom name for the server (defaults to server's name attribute or file name)",
),
] = None,
with_editable: Annotated[
Optional[Path],
typer.Option(
"--with-editable",
"-e",
help="Directory containing pyproject.toml to install in editable mode",
exists=True,
file_okay=False,
resolve_path=True,
),
] = None,
with_packages: Annotated[
list[str],
typer.Option(
"--with",
help="Additional packages to install",
),
] = [],
env_vars: Annotated[
list[str],
typer.Option(
"--env-var",
"-e",
help="Environment variables in KEY=VALUE format",
),
] = [],
env_file: Annotated[
Optional[Path],
typer.Option(
"--env-file",
"-f",
help="Load environment variables from a .env file",
exists=True,
file_okay=True,
dir_okay=False,
resolve_path=True,
),
] = None,
) -> None:
"""Install a FastMCP server in the Claude desktop app.
Environment variables are preserved once added and only updated if new values
are explicitly provided.
"""
file, server_object = _parse_file_path(file_spec)
logger.debug(
"Installing server",
extra={
"file": str(file),
"server_name": server_name,
"server_object": server_object,
"with_editable": str(with_editable) if with_editable else None,
"with_packages": with_packages,
},
)
if not claude.get_claude_config_path():
logger.error("Claude app not found")
sys.exit(1)
# Try to import server to get its name, but fall back to file name if dependencies missing
name = server_name
server = None
if not name:
try:
server = _import_server(file, server_object)
name = server.name
except (ImportError, ModuleNotFoundError) as e:
logger.debug(
"Could not import server (likely missing dependencies), using file name",
extra={"error": str(e)},
)
name = file.stem
# Get server dependencies if available
server_dependencies = getattr(server, "dependencies", []) if server else []
if server_dependencies:
with_packages = list(set(with_packages + server_dependencies))
# Process environment variables if provided
env_dict: Optional[Dict[str, str]] = None
if env_file or env_vars:
env_dict = {}
# Load from .env file if specified
if env_file:
try:
env_dict |= {
k: v
for k, v in dotenv.dotenv_values(env_file).items()
if v is not None
}
except Exception as e:
logger.error(f"Failed to load .env file: {e}")
sys.exit(1)
# Add command line environment variables
for env_var in env_vars:
key, value = _parse_env_var(env_var)
env_dict[key] = value
if claude.update_claude_config(
file_spec,
name,
with_editable=with_editable,
with_packages=with_packages,
env_vars=env_dict,
):
logger.info(f"Successfully installed {name} in Claude app")
else:
logger.error(f"Failed to install {name} in Claude app")
sys.exit(1)

21
src/fastmcp/exceptions.py
View file

@ -1,21 +0,0 @@
"""Custom exceptions for FastMCP."""
class FastMCPError(Exception):
"""Base error for FastMCP."""
class ValidationError(FastMCPError):
"""Error in validating parameters or return values."""
class ResourceError(FastMCPError):
"""Error in resource operations."""
class ToolError(FastMCPError):
"""Error in tool operations."""
class InvalidSignature(Exception):
"""Invalid signature for use with FastMCP."""

4
src/fastmcp/prompts/__init__.py
View file

@ -1,4 +0,0 @@
from .base import Prompt
from .manager import PromptManager
__all__ = ["Prompt", "PromptManager"]

165
src/fastmcp/prompts/base.py
View file

@ -1,165 +0,0 @@
"""Base classes for FastMCP prompts."""
import json
from typing import Any, Callable, Dict, Literal, Optional, Sequence, Awaitable
import inspect
from pydantic import BaseModel, Field, TypeAdapter, validate_call
from mcp.types import TextContent, ImageContent, EmbeddedResource
import pydantic_core
CONTENT_TYPES = TextContent | ImageContent | EmbeddedResource
class Message(BaseModel):
"""Base class for all prompt messages."""
role: Literal["user", "assistant"]
content: CONTENT_TYPES
def __init__(self, content: str | CONTENT_TYPES, **kwargs):
if isinstance(content, str):
content = TextContent(type="text", text=content)
super().__init__(content=content, **kwargs)
class UserMessage(Message):
"""A message from the user."""
role: Literal["user"] = "user"
def __init__(self, content: str | CONTENT_TYPES, **kwargs):
super().__init__(content=content, **kwargs)
class AssistantMessage(Message):
"""A message from the assistant."""
role: Literal["assistant"] = "assistant"
def __init__(self, content: str | CONTENT_TYPES, **kwargs):
super().__init__(content=content, **kwargs)
message_validator = TypeAdapter(UserMessage | AssistantMessage)
SyncPromptResult = (
str | Message | dict[str, Any] | Sequence[str | Message | dict[str, Any]]
)
PromptResult = SyncPromptResult | Awaitable[SyncPromptResult]
class PromptArgument(BaseModel):
"""An argument that can be passed to a prompt."""
name: str = Field(description="Name of the argument")
description: str | None = Field(
None, description="Description of what the argument does"
)
required: bool = Field(
default=False, description="Whether the argument is required"
)
class Prompt(BaseModel):
"""A prompt template that can be rendered with parameters."""
name: str = Field(description="Name of the prompt")
description: str | None = Field(
None, description="Description of what the prompt does"
)
arguments: list[PromptArgument] | None = Field(
None, description="Arguments that can be passed to the prompt"
)
fn: Callable = Field(exclude=True)
@classmethod
def from_function(
cls,
fn: Callable[..., PromptResult],
name: Optional[str] = None,
description: Optional[str] = None,
) -> "Prompt":
"""Create a Prompt from a function.
The function can return:
- A string (converted to a message)
- A Message object
- A dict (converted to a message)
- A sequence of any of the above
"""
func_name = name or fn.__name__
if func_name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
# Get schema from TypeAdapter - will fail if function isn't properly typed
parameters = TypeAdapter(fn).json_schema()
# Convert parameters to PromptArguments
arguments = []
if "properties" in parameters:
for param_name, param in parameters["properties"].items():
required = param_name in parameters.get("required", [])
arguments.append(
PromptArgument(
name=param_name,
description=param.get("description"),
required=required,
)
)
# ensure the arguments are properly cast
fn = validate_call(fn)
return cls(
name=func_name,
description=description or fn.__doc__ or "",
arguments=arguments,
fn=fn,
)
async def render(self, arguments: Optional[Dict[str, Any]] = None) -> list[Message]:
"""Render the prompt with arguments."""
# Validate required arguments
if self.arguments:
required = {arg.name for arg in self.arguments if arg.required}
provided = set(arguments or {})
missing = required - provided
if missing:
raise ValueError(f"Missing required arguments: {missing}")
try:
# Call function and check if result is a coroutine
result = self.fn(**(arguments or {}))
if inspect.iscoroutine(result):
result = await result
# Validate messages
if not isinstance(result, (list, tuple)):
result = [result]
# Convert result to messages
messages = []
for msg in result:
try:
if isinstance(msg, Message):
messages.append(msg)
elif isinstance(msg, dict):
msg = message_validator.validate_python(msg)
messages.append(msg)
elif isinstance(msg, str):
messages.append(
UserMessage(content=TextContent(type="text", text=msg))
)
else:
msg = json.dumps(pydantic_core.to_jsonable_python(msg))
messages.append(Message(role="user", content=msg))
except Exception:
raise ValueError(
f"Could not convert prompt result to message: {msg}"
)
return messages
except Exception as e:
raise ValueError(f"Error rendering prompt {self.name}: {e}")

50
src/fastmcp/prompts/manager.py
View file

@ -1,50 +0,0 @@
"""Prompt management functionality."""
from typing import Any, Dict, Optional
from fastmcp.prompts.base import Message, Prompt
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class PromptManager:
"""Manages FastMCP prompts."""
def __init__(self, warn_on_duplicate_prompts: bool = True):
self._prompts: Dict[str, Prompt] = {}
self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
def get_prompt(self, name: str) -> Optional[Prompt]:
"""Get prompt by name."""
return self._prompts.get(name)
def list_prompts(self) -> list[Prompt]:
"""List all registered prompts."""
return list(self._prompts.values())
def add_prompt(
self,
prompt: Prompt,
) -> Prompt:
"""Add a prompt to the manager."""
# Check for duplicates
existing = self._prompts.get(prompt.name)
if existing:
if self.warn_on_duplicate_prompts:
logger.warning(f"Prompt already exists: {prompt.name}")
return existing
self._prompts[prompt.name] = prompt
return prompt
async def render_prompt(
self, name: str, arguments: Optional[Dict[str, Any]] = None
) -> list[Message]:
"""Render a prompt by name with arguments."""
prompt = self.get_prompt(name)
if not prompt:
raise ValueError(f"Unknown prompt: {name}")
return await prompt.render(arguments)

36
src/fastmcp/prompts/prompt_manager.py
View file

@ -1,36 +0,0 @@
"""Prompt management functionality."""
from typing import Dict, Optional
from fastmcp.prompts.base import Prompt
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class PromptManager:
"""Manages FastMCP prompts."""
def __init__(self, warn_on_duplicate_prompts: bool = True):
self._prompts: Dict[str, Prompt] = {}
self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
def add_prompt(self, prompt: Prompt) -> Prompt:
"""Add a prompt to the manager."""
logger.debug(f"Adding prompt: {prompt.name}")
existing = self._prompts.get(prompt.name)
if existing:
if self.warn_on_duplicate_prompts:
logger.warning(f"Prompt already exists: {prompt.name}")
return existing
self._prompts[prompt.name] = prompt
return prompt
def get_prompt(self, name: str) -> Optional[Prompt]:
"""Get prompt by name."""
return self._prompts.get(name)
def list_prompts(self) -> list[Prompt]:
"""List all registered prompts."""
return list(self._prompts.values())

23
src/fastmcp/resources/__init__.py
View file

@ -1,23 +0,0 @@
from .base import Resource
from .types import (
TextResource,
BinaryResource,
FunctionResource,
FileResource,
HttpResource,
DirectoryResource,
)
from .templates import ResourceTemplate
from .resource_manager import ResourceManager
__all__ = [
"Resource",
"TextResource",
"BinaryResource",
"FunctionResource",
"FileResource",
"HttpResource",
"DirectoryResource",
"ResourceTemplate",
"ResourceManager",
]

48
src/fastmcp/resources/base.py
View file

@ -1,48 +0,0 @@
"""Base classes and interfaces for FastMCP resources."""
import abc
from typing import Union, Annotated
from pydantic import (
AnyUrl,
BaseModel,
ConfigDict,
Field,
UrlConstraints,
ValidationInfo,
field_validator,
)
class Resource(BaseModel, abc.ABC):
"""Base class for all resources."""
model_config = ConfigDict(validate_default=True)
uri: Annotated[AnyUrl, UrlConstraints(host_required=False)] = Field(
default=..., description="URI of the resource"
)
name: str | None = Field(description="Name of the resource", default=None)
description: str | None = Field(
description="Description of the resource", default=None
)
mime_type: str = Field(
default="text/plain",
description="MIME type of the resource content",
pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$",
)
@field_validator("name", mode="before")
@classmethod
def set_default_name(cls, name: str | None, info: ValidationInfo) -> str:
"""Set default name from URI if not provided."""
if name:
return name
if uri := info.data.get("uri"):
return str(uri)
raise ValueError("Either name or uri must be provided")
@abc.abstractmethod
async def read(self) -> Union[str, bytes]:
"""Read the resource content."""
pass

94
src/fastmcp/resources/resource_manager.py
View file

@ -1,94 +0,0 @@
"""Resource manager functionality."""
from typing import Callable, Dict, Optional, Union
from pydantic import AnyUrl
from fastmcp.resources.base import Resource
from fastmcp.resources.templates import ResourceTemplate
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class ResourceManager:
"""Manages FastMCP resources."""
def __init__(self, warn_on_duplicate_resources: bool = True):
self._resources: Dict[str, Resource] = {}
self._templates: Dict[str, ResourceTemplate] = {}
self.warn_on_duplicate_resources = warn_on_duplicate_resources
def add_resource(self, resource: Resource) -> Resource:
"""Add a resource to the manager.
Args:
resource: A Resource instance to add
Returns:
The added resource. If a resource with the same URI already exists,
returns the existing resource.
"""
logger.debug(
"Adding resource",
extra={
"uri": resource.uri,
"type": type(resource).__name__,
"name": resource.name,
},
)
existing = self._resources.get(str(resource.uri))
if existing:
if self.warn_on_duplicate_resources:
logger.warning(f"Resource already exists: {resource.uri}")
return existing
self._resources[str(resource.uri)] = resource
return resource
def add_template(
self,
fn: Callable,
uri_template: str,
name: Optional[str] = None,
description: Optional[str] = None,
mime_type: Optional[str] = None,
) -> ResourceTemplate:
"""Add a template from a function."""
template = ResourceTemplate.from_function(
fn,
uri_template=uri_template,
name=name,
description=description,
mime_type=mime_type,
)
self._templates[template.uri_template] = template
return template
async def get_resource(self, uri: Union[AnyUrl, str]) -> Optional[Resource]:
"""Get resource by URI, checking concrete resources first, then templates."""
uri_str = str(uri)
logger.debug("Getting resource", extra={"uri": uri_str})
# First check concrete resources
if resource := self._resources.get(uri_str):
return resource
# Then check templates
for template in self._templates.values():
if params := template.matches(uri_str):
try:
return await template.create_resource(uri_str, params)
except Exception as e:
raise ValueError(f"Error creating resource from template: {e}")
raise ValueError(f"Unknown resource: {uri}")
def list_resources(self) -> list[Resource]:
"""List all registered resources."""
logger.debug("Listing resources", extra={"count": len(self._resources)})
return list(self._resources.values())
def list_templates(self) -> list[ResourceTemplate]:
"""List all registered templates."""
logger.debug("Listing templates", extra={"count": len(self._templates)})
return list(self._templates.values())

80
src/fastmcp/resources/templates.py
View file

@ -1,80 +0,0 @@
"""Resource template functionality."""
import inspect
import re
from typing import Any, Callable, Dict, Optional
from pydantic import BaseModel, Field, TypeAdapter, validate_call
from fastmcp.resources.types import FunctionResource, Resource
class ResourceTemplate(BaseModel):
"""A template for dynamically creating resources."""
uri_template: str = Field(
description="URI template with parameters (e.g. weather://{city}/current)"
)
name: str = Field(description="Name of the resource")
description: str | None = Field(description="Description of what the resource does")
mime_type: str = Field(
default="text/plain", description="MIME type of the resource content"
)
fn: Callable = Field(exclude=True)
parameters: dict = Field(description="JSON schema for function parameters")
@classmethod
def from_function(
cls,
fn: Callable,
uri_template: str,
name: Optional[str] = None,
description: Optional[str] = None,
mime_type: Optional[str] = None,
) -> "ResourceTemplate":
"""Create a template from a function."""
func_name = name or fn.__name__
if func_name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
# Get schema from TypeAdapter - will fail if function isn't properly typed
parameters = TypeAdapter(fn).json_schema()
# ensure the arguments are properly cast
fn = validate_call(fn)
return cls(
uri_template=uri_template,
name=func_name,
description=description or fn.__doc__ or "",
mime_type=mime_type or "text/plain",
fn=fn,
parameters=parameters,
)
def matches(self, uri: str) -> Optional[Dict[str, Any]]:
"""Check if URI matches template and extract parameters."""
# Convert template to regex pattern
pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
match = re.match(f"^{pattern}$", uri)
if match:
return match.groupdict()
return None
async def create_resource(self, uri: str, params: Dict[str, Any]) -> Resource:
"""Create a resource from the template with the given parameters."""
try:
# Call function and check if result is a coroutine
result = self.fn(**params)
if inspect.iscoroutine(result):
result = await result
return FunctionResource(
uri=uri, # type: ignore
name=self.name,
description=self.description,
mime_type=self.mime_type,
fn=lambda: result, # Capture result in closure
)
except Exception as e:
raise ValueError(f"Error creating resource from template: {e}")

180
src/fastmcp/resources/types.py
View file

@ -1,180 +0,0 @@
"""Concrete resource implementations."""
import asyncio
import json
from pathlib import Path
from typing import Any, Callable, Union
import httpx
import pydantic.json
import pydantic_core
from pydantic import Field, ValidationInfo
from fastmcp.resources.base import Resource
class TextResource(Resource):
"""A resource that reads from a string."""
text: str = Field(description="Text content of the resource")
async def read(self) -> str:
"""Read the text content."""
return self.text
class BinaryResource(Resource):
"""A resource that reads from bytes."""
data: bytes = Field(description="Binary content of the resource")
async def read(self) -> bytes:
"""Read the binary content."""
return self.data
class FunctionResource(Resource):
"""A resource that defers data loading by wrapping a function.
The function is only called when the resource is read, allowing for lazy loading
of potentially expensive data. This is particularly useful when listing resources,
as the function won't be called until the resource is actually accessed.
The function can return:
- str for text content (default)
- bytes for binary content
- other types will be converted to JSON
"""
fn: Callable[[], Any] = Field(exclude=True)
async def read(self) -> Union[str, bytes]:
"""Read the resource by calling the wrapped function."""
try:
result = self.fn()
if isinstance(result, Resource):
return await result.read()
if isinstance(result, bytes):
return result
if isinstance(result, str):
return result
try:
return json.dumps(pydantic_core.to_jsonable_python(result))
except (TypeError, pydantic_core.PydanticSerializationError):
# If JSON serialization fails, try str()
return str(result)
except Exception as e:
raise ValueError(f"Error reading resource {self.uri}: {e}")
class FileResource(Resource):
"""A resource that reads from a file.
Set is_binary=True to read file as binary data instead of text.
"""
path: Path = Field(description="Path to the file")
is_binary: bool = Field(
default=False,
description="Whether to read the file as binary data",
)
mime_type: str = Field(
default="text/plain",
description="MIME type of the resource content",
)
@pydantic.field_validator("path")
@classmethod
def validate_absolute_path(cls, path: Path) -> Path:
"""Ensure path is absolute."""
if not path.is_absolute():
raise ValueError("Path must be absolute")
return path
@pydantic.field_validator("is_binary")
@classmethod
def set_binary_from_mime_type(cls, is_binary: bool, info: ValidationInfo) -> bool:
"""Set is_binary based on mime_type if not explicitly set."""
if is_binary:
return True
mime_type = info.data.get("mime_type", "text/plain")
return not mime_type.startswith("text/")
async def read(self) -> Union[str, bytes]:
"""Read the file content."""
try:
if self.is_binary:
return await asyncio.to_thread(self.path.read_bytes)
return await asyncio.to_thread(self.path.read_text)
except Exception as e:
raise ValueError(f"Error reading file {self.path}: {e}")
class HttpResource(Resource):
"""A resource that reads from an HTTP endpoint."""
url: str = Field(description="URL to fetch content from")
mime_type: str | None = Field(
default="application/json", description="MIME type of the resource content"
)
async def read(self) -> Union[str, bytes]:
"""Read the HTTP content."""
async with httpx.AsyncClient() as client:
response = await client.get(self.url)
response.raise_for_status()
return response.text
class DirectoryResource(Resource):
"""A resource that lists files in a directory."""
path: Path = Field(description="Path to the directory")
recursive: bool = Field(
default=False, description="Whether to list files recursively"
)
pattern: str | None = Field(
default=None, description="Optional glob pattern to filter files"
)
mime_type: str | None = Field(
default="application/json", description="MIME type of the resource content"
)
@pydantic.field_validator("path")
@classmethod
def validate_absolute_path(cls, path: Path) -> Path:
"""Ensure path is absolute."""
if not path.is_absolute():
raise ValueError("Path must be absolute")
return path
def list_files(self) -> list[Path]:
"""List files in the directory."""
if not self.path.exists():
raise FileNotFoundError(f"Directory not found: {self.path}")
if not self.path.is_dir():
raise NotADirectoryError(f"Not a directory: {self.path}")
try:
if self.pattern:
return (
list(self.path.glob(self.pattern))
if not self.recursive
else list(self.path.rglob(self.pattern))
)
return (
list(self.path.glob("*"))
if not self.recursive
else list(self.path.rglob("*"))
)
except Exception as e:
raise ValueError(f"Error listing directory {self.path}: {e}")
async def read(self) -> str: # Always returns JSON string
"""Read the directory listing."""
try:
files = await asyncio.to_thread(self.list_files)
file_list = [str(f.relative_to(self.path)) for f in files if f.is_file()]
return json.dumps({"files": file_list}, indent=2)
except Exception as e:
raise ValueError(f"Error reading directory {self.path}: {e}")

670
src/fastmcp/server.py
View file

@ -1,671 +1,11 @@
"""FastMCP - A more ergonomic interface for MCP servers."""
from typing import Any
import asyncio
import functools
import inspect
import json
import re
from itertools import chain
from typing import Any, Callable, Dict, Literal, Sequence, TypeVar, ParamSpec
import pydantic_core
from pydantic import Field
import uvicorn
from mcp.server import Server as MCPServer
from mcp.server.sse import SseServerTransport
from mcp.server.stdio import stdio_server
from mcp.shared.context import RequestContext
from mcp.types import (
EmbeddedResource,
GetPromptResult,
ImageContent,
TextContent,
)
from mcp.types import (
Prompt as MCPPrompt,
PromptArgument as MCPPromptArgument,
)
from mcp.types import (
Resource as MCPResource,
)
from mcp.types import (
ResourceTemplate as MCPResourceTemplate,
)
from mcp.types import (
Tool as MCPTool,
)
from pydantic import BaseModel
from pydantic.networks import AnyUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
from fastmcp.exceptions import ResourceError
from fastmcp.prompts import Prompt, PromptManager
from fastmcp.prompts.base import PromptResult
from fastmcp.resources import FunctionResource, Resource, ResourceManager
from fastmcp.tools import ToolManager
from fastmcp.utilities.logging import configure_logging, get_logger
from fastmcp.utilities.types import Image
import mcp.server.fastmcp
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
P = ParamSpec("P")
R = TypeVar("R")
R_PromptResult = TypeVar("R_PromptResult", bound=PromptResult)
class Settings(BaseSettings):
"""FastMCP server settings.
All settings can be configured via environment variables with the prefix FASTMCP_.
For example, FASTMCP_DEBUG=true will set debug=True.
"""
model_config: SettingsConfigDict = SettingsConfigDict(
env_prefix="FASTMCP_",
env_file=".env",
extra="ignore",
)
# Server settings
debug: bool = False
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
# HTTP settings
host: str = "0.0.0.0"
port: int = 8000
# resource settings
warn_on_duplicate_resources: bool = True
# tool settings
warn_on_duplicate_tools: bool = True
# prompt settings
warn_on_duplicate_prompts: bool = True
dependencies: list[str] = Field(
default_factory=list,
description="List of dependencies to install in the server environment",
)
class FastMCP:
class FastMCP(mcp.server.fastmcp.FastMCP):
def __init__(self, name: str | None = None, **settings: Any):
self.settings = Settings(**settings)
self._mcp_server = MCPServer(name=name or "FastMCP")
self._tool_manager = ToolManager(
warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools
)
self._resource_manager = ResourceManager(
warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources
)
self._prompt_manager = PromptManager(
warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts
)
self.dependencies = self.settings.dependencies
# Set up MCP protocol handlers
self._setup_handlers()
# Configure logging
configure_logging(self.settings.log_level)
@property
def name(self) -> str:
return self._mcp_server.name
def run(self, transport: Literal["stdio", "sse"] = "stdio") -> None:
"""Run the FastMCP server. Note this is a synchronous function.
Args:
transport: Transport protocol to use ("stdio" or "sse")
"""
TRANSPORTS = Literal["stdio", "sse"]
if transport not in TRANSPORTS.__args__: # type: ignore
raise ValueError(f"Unknown transport: {transport}")
if transport == "stdio":
asyncio.run(self.run_stdio_async())
else: # transport == "sse"
asyncio.run(self.run_sse_async())
def _setup_handlers(self) -> None:
"""Set up core MCP protocol handlers."""
self._mcp_server.list_tools()(self.list_tools)
self._mcp_server.call_tool()(self.call_tool)
self._mcp_server.list_resources()(self.list_resources)
self._mcp_server.read_resource()(self.read_resource)
self._mcp_server.list_prompts()(self.list_prompts)
self._mcp_server.get_prompt()(self.get_prompt)
# TODO: This has not been added to MCP yet, see https://github.com/jlowin/fastmcp/issues/10
# self._mcp_server.list_resource_templates()(self.list_resource_templates)
async def list_tools(self) -> list[MCPTool]:
"""List all available tools."""
tools = self._tool_manager.list_tools()
return [
MCPTool(
name=info.name,
description=info.description,
inputSchema=info.parameters,
)
for info in tools
]
def get_context(self) -> "Context":
"""
Returns a Context object. Note that the context will only be valid
during a request; outside a request, most methods will error.
"""
try:
request_context = self._mcp_server.request_context
except LookupError:
request_context = None
return Context(request_context=request_context, fastmcp=self)
async def call_tool(
self, name: str, arguments: dict
) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""Call a tool by name with arguments."""
context = self.get_context()
result = await self._tool_manager.call_tool(name, arguments, context=context)
converted_result = _convert_to_content(result)
return converted_result
async def list_resources(self) -> list[MCPResource]:
"""List all available resources."""
resources = self._resource_manager.list_resources()
return [
MCPResource(
uri=resource.uri,
name=resource.name or "",
description=resource.description,
mimeType=resource.mime_type,
)
for resource in resources
]
async def list_resource_templates(self) -> list[MCPResourceTemplate]:
templates = self._resource_manager.list_templates()
return [
MCPResourceTemplate(
uriTemplate=template.uri_template,
name=template.name,
description=template.description,
)
for template in templates
]
async def read_resource(self, uri: AnyUrl | str) -> str | bytes:
"""Read a resource by URI."""
resource = await self._resource_manager.get_resource(uri)
if not resource:
raise ResourceError(f"Unknown resource: {uri}")
try:
return await resource.read()
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
raise ResourceError(str(e))
def add_tool(
self,
fn: Callable,
name: str | None = None,
description: str | None = None,
) -> None:
"""Add a tool to the server.
The tool function can optionally request a Context object by adding a parameter
with the Context type annotation. See the @tool decorator for examples.
Args:
fn: The function to register as a tool
name: Optional name for the tool (defaults to function name)
description: Optional description of what the tool does
"""
self._tool_manager.add_tool(fn, name=name, description=description)
def tool(
self, name: str | None = None, description: str | None = None
) -> Callable[[Callable[P, R]], Callable[P, R]]:
"""Decorator to register a tool.
Tools can optionally request a Context object by adding a parameter with the Context type annotation.
The context provides access to MCP capabilities like logging, progress reporting, and resource access.
Args:
name: Optional name for the tool (defaults to function name)
description: Optional description of what the tool does
Example:
@server.tool()
def my_tool(x: int) -> str:
return str(x)
@server.tool()
def tool_with_context(x: int, ctx: Context) -> str:
ctx.info(f"Processing {x}")
return str(x)
@server.tool()
async def async_tool(x: int, context: Context) -> str:
await context.report_progress(50, 100)
return str(x)
"""
# Check if user passed function directly instead of calling decorator
if callable(name):
raise TypeError(
"The @tool decorator was used incorrectly. "
"Did you forget to call it? Use @tool() instead of @tool"
)
def decorator(fn: Callable[P, R]) -> Callable[P, R]:
self.add_tool(fn, name=name, description=description)
return fn
return decorator
def add_resource(self, resource: Resource) -> None:
"""Add a resource to the server.
Args:
resource: A Resource instance to add
"""
self._resource_manager.add_resource(resource)
def resource(
self,
uri: str,
*,
name: str | None = None,
description: str | None = None,
mime_type: str | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
"""Decorator to register a function as a resource.
The function will be called when the resource is read to generate its content.
The function can return:
- str for text content
- bytes for binary content
- other types will be converted to JSON
If the URI contains parameters (e.g. "resource://{param}") or the function
has parameters, it will be registered as a template resource.
Args:
uri: URI for the resource (e.g. "resource://my-resource" or "resource://{param}")
name: Optional name for the resource
description: Optional description of the resource
mime_type: Optional MIME type for the resource
Example:
@server.resource("resource://my-resource")
def get_data() -> str:
return "Hello, world!"
@server.resource("resource://{city}/weather")
def get_weather(city: str) -> str:
return f"Weather for {city}"
"""
# Check if user passed function directly instead of calling decorator
if callable(uri):
raise TypeError(
"The @resource decorator was used incorrectly. "
"Did you forget to call it? Use @resource('uri') instead of @resource"
)
def decorator(fn: Callable[P, R]) -> Callable[P, R]:
@functools.wraps(fn)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
return fn(*args, **kwargs)
# Check if this should be a template
has_uri_params = "{" in uri and "}" in uri
has_func_params = bool(inspect.signature(fn).parameters)
if has_uri_params or has_func_params:
# Validate that URI params match function params
uri_params = set(re.findall(r"{(\w+)}", uri))
func_params = set(inspect.signature(fn).parameters.keys())
if uri_params != func_params:
raise ValueError(
f"Mismatch between URI parameters {uri_params} "
f"and function parameters {func_params}"
)
# Register as template
self._resource_manager.add_template(
wrapper,
uri_template=uri,
name=name,
description=description,
mime_type=mime_type or "text/plain",
)
else:
# Register as regular resource
resource = FunctionResource(
uri=AnyUrl(uri),
name=name,
description=description,
mime_type=mime_type or "text/plain",
fn=wrapper,
)
self.add_resource(resource)
return wrapper
return decorator
def add_prompt(self, prompt: Prompt) -> None:
"""Add a prompt to the server.
Args:
prompt: A Prompt instance to add
"""
self._prompt_manager.add_prompt(prompt)
def prompt(
self, name: str | None = None, description: str | None = None
) -> Callable[[Callable[P, R_PromptResult]], Callable[P, R_PromptResult]]:
"""Decorator to register a prompt.
Args:
name: Optional name for the prompt (defaults to function name)
description: Optional description of what the prompt does
Example:
@server.prompt()
def analyze_table(table_name: str) -> list[Message]:
schema = read_table_schema(table_name)
return [
{
"role": "user",
"content": f"Analyze this schema:\n{schema}"
}
]
@server.prompt()
async def analyze_file(path: str) -> list[Message]:
content = await read_file(path)
return [
{
"role": "user",
"content": {
"type": "resource",
"resource": {
"uri": f"file://{path}",
"text": content
}
}
}
]
"""
# Check if user passed function directly instead of calling decorator
if callable(name):
raise TypeError(
"The @prompt decorator was used incorrectly. "
"Did you forget to call it? Use @prompt() instead of @prompt"
)
def decorator(func: Callable[P, R_PromptResult]) -> Callable[P, R_PromptResult]:
prompt = Prompt.from_function(func, name=name, description=description)
self.add_prompt(prompt)
return func
return decorator
async def run_stdio_async(self) -> None:
"""Run the server using stdio transport."""
async with stdio_server() as (read_stream, write_stream):
await self._mcp_server.run(
read_stream,
write_stream,
self._mcp_server.create_initialization_options(),
)
async def run_sse_async(self) -> None:
"""Run the server using SSE transport."""
from starlette.applications import Starlette
from starlette.routing import Route, Mount
sse = SseServerTransport("/messages/")
async def handle_sse(request):
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await self._mcp_server.run(
streams[0],
streams[1],
self._mcp_server.create_initialization_options(),
)
starlette_app = Starlette(
debug=self.settings.debug,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
)
config = uvicorn.Config(
starlette_app,
host=self.settings.host,
port=self.settings.port,
log_level=self.settings.log_level.lower(),
)
server = uvicorn.Server(config)
await server.serve()
async def list_prompts(self) -> list[MCPPrompt]:
"""List all available prompts."""
prompts = self._prompt_manager.list_prompts()
return [
MCPPrompt(
name=prompt.name,
description=prompt.description,
arguments=[
MCPPromptArgument(
name=arg.name,
description=arg.description,
required=arg.required,
)
for arg in (prompt.arguments or [])
],
)
for prompt in prompts
]
async def get_prompt(
self, name: str, arguments: Dict[str, Any] | None = None
) -> GetPromptResult:
"""Get a prompt by name with arguments."""
try:
messages = await self._prompt_manager.render_prompt(name, arguments)
return GetPromptResult(messages=pydantic_core.to_jsonable_python(messages))
except Exception as e:
logger.error(f"Error getting prompt {name}: {e}")
raise ValueError(str(e))
def _convert_to_content(
result: Any,
) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""Convert a result to a sequence of content objects."""
if result is None:
return []
if isinstance(result, (TextContent, ImageContent, EmbeddedResource)):
return [result]
if isinstance(result, Image):
return [result.to_image_content()]
if isinstance(result, (list, tuple)):
return list(chain.from_iterable(_convert_to_content(item) for item in result))
if not isinstance(result, str):
try:
result = json.dumps(pydantic_core.to_jsonable_python(result))
except Exception:
result = str(result)
return [TextContent(type="text", text=result)]
class Context(BaseModel):
"""Context object providing access to MCP capabilities.
This provides a cleaner interface to MCP's RequestContext functionality.
It gets injected into tool and resource functions that request it via type hints.
To use context in a tool function, add a parameter with the Context type annotation:
```python
@server.tool()
def my_tool(x: int, ctx: Context) -> str:
# Log messages to the client
ctx.info(f"Processing {x}")
ctx.debug("Debug info")
ctx.warning("Warning message")
ctx.error("Error message")
# Report progress
ctx.report_progress(50, 100)
# Access resources
data = ctx.read_resource("resource://data")
# Get request info
request_id = ctx.request_id
client_id = ctx.client_id
return str(x)
```
The context parameter name can be anything as long as it's annotated with Context.
The context is optional - tools that don't need it can omit the parameter.
"""
_request_context: RequestContext | None
_fastmcp: FastMCP | None
def __init__(
self,
*,
request_context: RequestContext | None = None,
fastmcp: FastMCP | None = None,
**kwargs: Any,
):
super().__init__(**kwargs)
self._request_context = request_context
self._fastmcp = fastmcp
@property
def fastmcp(self) -> FastMCP:
"""Access to the FastMCP server."""
if self._fastmcp is None:
raise ValueError("Context is not available outside of a request")
return self._fastmcp
@property
def request_context(self) -> RequestContext:
"""Access to the underlying request context."""
if self._request_context is None:
raise ValueError("Context is not available outside of a request")
return self._request_context
async def report_progress(
self, progress: float, total: float | None = None
) -> None:
"""Report progress for the current operation.
Args:
progress: Current progress value e.g. 24
total: Optional total value e.g. 100
"""
progress_token = (
self.request_context.meta.progressToken
if self.request_context.meta
else None
)
if not progress_token:
return
await self.request_context.session.send_progress_notification(
progress_token=progress_token, progress=progress, total=total
)
async def read_resource(self, uri: str | AnyUrl) -> str | bytes:
"""Read a resource by URI.
Args:
uri: Resource URI to read
Returns:
The resource content as either text or bytes
"""
assert (
self._fastmcp is not None
), "Context is not available outside of a request"
return await self._fastmcp.read_resource(uri)
def log(
self,
level: Literal["debug", "info", "warning", "error"],
message: str,
*,
logger_name: str | None = None,
) -> None:
"""Send a log message to the client.
Args:
level: Log level (debug, info, warning, error)
message: Log message
logger_name: Optional logger name
**extra: Additional structured data to include
"""
self.request_context.session.send_log_message(
level=level, data=message, logger=logger_name
)
@property
def client_id(self) -> str | None:
"""Get the client ID if available."""
return (
getattr(self.request_context.meta, "client_id", None)
if self.request_context.meta
else None
)
@property
def request_id(self) -> str:
"""Get the unique ID for this request."""
return str(self.request_context.request_id)
@property
def session(self):
"""Access to the underlying session for advanced usage."""
return self.request_context.session
# Convenience methods for common log levels
def debug(self, message: str, **extra: Any) -> None:
"""Send a debug log message."""
self.log("debug", message, **extra)
def info(self, message: str, **extra: Any) -> None:
"""Send an info log message."""
self.log("info", message, **extra)
def warning(self, message: str, **extra: Any) -> None:
"""Send a warning log message."""
self.log("warning", message, **extra)
def error(self, message: str, **extra: Any) -> None:
"""Send an error log message."""
self.log("error", message, **extra)
super().__init__(name=name or "FastMCP", **settings)

41
src/fastmcp/settings.py Normal file
View file

@ -0,0 +1,41 @@
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Literal
class Settings(BaseSettings):
"""FastMCP server settings.
All settings can be configured via environment variables with the prefix FASTMCP_.
For example, FASTMCP_DEBUG=true will set debug=True.
"""
model_config: SettingsConfigDict = SettingsConfigDict(
env_prefix="FASTMCP_",
env_file=".env",
extra="ignore",
)
# Server settings
debug: bool = False
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
# HTTP settings
host: str = "0.0.0.0"
port: int = 8000
# resource settings
warn_on_duplicate_resources: bool = True
# tool settings
warn_on_duplicate_tools: bool = True
# prompt settings
warn_on_duplicate_prompts: bool = True
dependencies: list[str] = Field(
default_factory=list,
description="List of dependencies to install in the server environment",
)

4
src/fastmcp/tools/__init__.py
View file

@ -1,4 +0,0 @@
from .base import Tool
from .tool_manager import ToolManager
__all__ = ["Tool", "ToolManager"]

83
src/fastmcp/tools/base.py
View file

@ -1,83 +0,0 @@
import fastmcp
from fastmcp.exceptions import ToolError
from fastmcp.utilities.func_metadata import func_metadata, FuncMetadata
from pydantic import BaseModel, Field
import inspect
from typing import TYPE_CHECKING, Any, Callable, Optional
if TYPE_CHECKING:
from fastmcp.server import Context
class Tool(BaseModel):
"""Internal tool registration info."""
fn: Callable = Field(exclude=True)
name: str = Field(description="Name of the tool")
description: str = Field(description="Description of what the tool does")
parameters: dict = Field(description="JSON schema for tool parameters")
fn_metadata: FuncMetadata = Field(
description="Metadata about the function including a pydantic model for tool arguments"
)
is_async: bool = Field(description="Whether the tool is async")
context_kwarg: Optional[str] = Field(
None, description="Name of the kwarg that should receive context"
)
@classmethod
def from_function(
cls,
fn: Callable,
name: Optional[str] = None,
description: Optional[str] = None,
context_kwarg: Optional[str] = None,
) -> "Tool":
"""Create a Tool from a function."""
func_name = name or fn.__name__
if func_name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
func_doc = description or fn.__doc__ or ""
is_async = inspect.iscoroutinefunction(fn)
# Find context parameter if it exists
if context_kwarg is None:
sig = inspect.signature(fn)
for param_name, param in sig.parameters.items():
if param.annotation is fastmcp.Context:
context_kwarg = param_name
break
func_arg_metadata = func_metadata(
fn,
skip_names=[context_kwarg] if context_kwarg is not None else [],
)
parameters = func_arg_metadata.arg_model.model_json_schema()
return cls(
fn=fn,
name=func_name,
description=func_doc,
parameters=parameters,
fn_metadata=func_arg_metadata,
is_async=is_async,
context_kwarg=context_kwarg,
)
async def run(self, arguments: dict, context: Optional["Context"] = None) -> Any:
"""Run the tool with arguments."""
try:
return await self.fn_metadata.call_fn_with_arg_validation(
self.fn,
self.is_async,
arguments,
{self.context_kwarg: context}
if self.context_kwarg is not None
else None,
)
except Exception as e:
raise ToolError(f"Error executing tool {self.name}: {e}") from e

55
src/fastmcp/tools/tool_manager.py
View file

@ -1,55 +0,0 @@
from fastmcp.exceptions import ToolError
from fastmcp.tools.base import Tool
from typing import Any, Callable, Dict, Optional, TYPE_CHECKING
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
from fastmcp.server import Context
logger = get_logger(__name__)
class ToolManager:
"""Manages FastMCP tools."""
def __init__(self, warn_on_duplicate_tools: bool = True):
self._tools: Dict[str, Tool] = {}
self.warn_on_duplicate_tools = warn_on_duplicate_tools
def get_tool(self, name: str) -> Optional[Tool]:
"""Get tool by name."""
return self._tools.get(name)
def list_tools(self) -> list[Tool]:
"""List all registered tools."""
return list(self._tools.values())
def add_tool(
self,
fn: Callable,
name: Optional[str] = None,
description: Optional[str] = None,
) -> Tool:
"""Add a tool to the server."""
tool = Tool.from_function(fn, name=name, description=description)
existing = self._tools.get(tool.name)
if existing:
if self.warn_on_duplicate_tools:
logger.warning(f"Tool already exists: {tool.name}")
return existing
self._tools[tool.name] = tool
return tool
async def call_tool(
self, name: str, arguments: dict, context: Optional["Context"] = None
) -> Any:
"""Call a tool by name with arguments."""
tool = self.get_tool(name)
if not tool:
raise ToolError(f"Unknown tool: {name}")
return await tool.run(arguments, context=context)

200
src/fastmcp/utilities/func_metadata.py
View file

@ -1,200 +0,0 @@
import inspect
from collections.abc import Callable, Sequence, Awaitable
from typing import (
Annotated,
Any,
Dict,
ForwardRef,
)
from pydantic import Field
from fastmcp.exceptions import InvalidSignature
from pydantic._internal._typing_extra import eval_type_lenient
import json
from pydantic import BaseModel
from pydantic.fields import FieldInfo
from pydantic import ConfigDict, create_model
from pydantic import WithJsonSchema
from pydantic_core import PydanticUndefined
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class ArgModelBase(BaseModel):
"""A model representing the arguments to a function."""
def model_dump_one_level(self) -> dict[str, Any]:
"""Return a dict of the model's fields, one level deep.
That is, sub-models etc are not dumped - they are kept as pydantic models.
"""
kwargs: dict[str, Any] = {}
for field_name in self.model_fields.keys():
kwargs[field_name] = getattr(self, field_name)
return kwargs
model_config = ConfigDict(
arbitrary_types_allowed=True,
)
class FuncMetadata(BaseModel):
arg_model: Annotated[type[ArgModelBase], WithJsonSchema(None)]
# We can add things in the future like
# - Maybe some args are excluded from attempting to parse from JSON
# - Maybe some args are special (like context) for dependency injection
async def call_fn_with_arg_validation(
self,
fn: Callable[..., Any] | Awaitable[Any],
fn_is_async: bool,
arguments_to_validate: dict[str, Any],
arguments_to_pass_directly: dict[str, Any] | None,
) -> Any:
"""Call the given function with arguments validated and injected.
Arguments are first attempted to be parsed from JSON, then validated against
the argument model, before being passed to the function.
"""
arguments_pre_parsed = self.pre_parse_json(arguments_to_validate)
arguments_parsed_model = self.arg_model.model_validate(arguments_pre_parsed)
arguments_parsed_dict = arguments_parsed_model.model_dump_one_level()
arguments_parsed_dict |= arguments_to_pass_directly or {}
if fn_is_async:
if isinstance(fn, Awaitable):
return await fn
return await fn(**arguments_parsed_dict)
if isinstance(fn, Callable):
return fn(**arguments_parsed_dict)
raise TypeError("fn must be either Callable or Awaitable")
def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
"""Pre-parse data from JSON.
Return a dict with same keys as input but with values parsed from JSON
if appropriate.
This is to handle cases like `["a", "b", "c"]` being passed in as JSON inside
a string rather than an actual list. Claude desktop is prone to this - in fact
it seems incapable of NOT doing this. For sub-models, it tends to pass
dicts (JSON objects) as JSON strings, which can be pre-parsed here.
"""
new_data = data.copy() # Shallow copy
for field_name, field_info in self.arg_model.model_fields.items():
if field_name not in data.keys():
continue
if isinstance(data[field_name], str):
try:
pre_parsed = json.loads(data[field_name])
except json.JSONDecodeError:
continue # Not JSON - skip
if isinstance(pre_parsed, (str, int, float)):
# This is likely that the raw value is e.g. `"hello"` which we
# Should really be parsed as '"hello"' in Python - but if we parse
# it as JSON it'll turn into just 'hello'. So we skip it.
continue
new_data[field_name] = pre_parsed
assert new_data.keys() == data.keys()
return new_data
model_config = ConfigDict(
arbitrary_types_allowed=True,
)
def func_metadata(func: Callable, skip_names: Sequence[str] = ()) -> FuncMetadata:
"""Given a function, return metadata including a pydantic model representing its signature.
The use case for this is
```
meta = func_to_pyd(func)
validated_args = meta.arg_model.model_validate(some_raw_data_dict)
return func(**validated_args.model_dump_one_level())
```
**critically** it also provides pre-parse helper to attempt to parse things from JSON.
Args:
func: The function to convert to a pydantic model
skip_names: A list of parameter names to skip. These will not be included in
the model.
Returns:
A pydantic model representing the function's signature.
"""
sig = _get_typed_signature(func)
params = sig.parameters
dynamic_pydantic_model_params: dict[str, Any] = {}
globalns = getattr(func, "__globals__", {})
for param in params.values():
if param.name.startswith("_"):
raise InvalidSignature(
f"Parameter {param.name} of {func.__name__} may not start with an underscore"
)
if param.name in skip_names:
continue
annotation = param.annotation
# `x: None` / `x: None = None`
if annotation is None:
annotation = Annotated[
None,
Field(
default=param.default
if param.default is not inspect.Parameter.empty
else PydanticUndefined
),
]
# Untyped field
if annotation is inspect.Parameter.empty:
annotation = Annotated[
Any,
Field(),
# 🤷
WithJsonSchema({"title": param.name, "type": "string"}),
]
field_info = FieldInfo.from_annotated_attribute(
_get_typed_annotation(annotation, globalns),
param.default
if param.default is not inspect.Parameter.empty
else PydanticUndefined,
)
dynamic_pydantic_model_params[param.name] = (field_info.annotation, field_info)
continue
arguments_model = create_model(
f"{func.__name__}Arguments",
**dynamic_pydantic_model_params,
__base__=ArgModelBase,
)
resp = FuncMetadata(arg_model=arguments_model)
return resp
def _get_typed_annotation(annotation: Any, globalns: Dict[str, Any]) -> Any:
if isinstance(annotation, str):
annotation = ForwardRef(annotation)
annotation = eval_type_lenient(annotation, globalns, globalns)
return annotation
def _get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
"""Get function signature while evaluating forward references"""
signature = inspect.signature(call)
globalns = getattr(call, "__globals__", {})
typed_params = [
inspect.Parameter(
name=param.name,
kind=param.kind,
default=param.default,
annotation=_get_typed_annotation(param.annotation, globalns),
)
for param in signature.parameters.values()
]
typed_signature = inspect.Signature(typed_params)
return typed_signature

0
tests/prompts/__init__.py
View file

194
tests/prompts/test_base.py
View file

@ -1,194 +0,0 @@
from pydantic import FileUrl
import pytest
from fastmcp.prompts.base import (
Prompt,
UserMessage,
TextContent,
AssistantMessage,
Message,
)
from mcp.types import EmbeddedResource, TextResourceContents
class TestRenderPrompt:
async def test_basic_fn(self):
def fn() -> str:
return "Hello, world!"
prompt = Prompt.from_function(fn)
assert await prompt.render() == [
UserMessage(content=TextContent(type="text", text="Hello, world!"))
]
async def test_async_fn(self):
async def fn() -> str:
return "Hello, world!"
prompt = Prompt.from_function(fn)
assert await prompt.render() == [
UserMessage(content=TextContent(type="text", text="Hello, world!"))
]
async def test_fn_with_args(self):
async def fn(name: str, age: int = 30) -> str:
return f"Hello, {name}! You're {age} years old."
prompt = Prompt.from_function(fn)
assert await prompt.render(arguments=dict(name="World")) == [
UserMessage(
content=TextContent(
type="text", text="Hello, World! You're 30 years old."
)
)
]
async def test_fn_with_invalid_kwargs(self):
async def fn(name: str, age: int = 30) -> str:
return f"Hello, {name}! You're {age} years old."
prompt = Prompt.from_function(fn)
with pytest.raises(ValueError):
await prompt.render(arguments=dict(age=40))
async def test_fn_returns_message(self):
async def fn() -> UserMessage:
return UserMessage(content="Hello, world!")
prompt = Prompt.from_function(fn)
assert await prompt.render() == [
UserMessage(content=TextContent(type="text", text="Hello, world!"))
]
async def test_fn_returns_assistant_message(self):
async def fn() -> AssistantMessage:
return AssistantMessage(
content=TextContent(type="text", text="Hello, world!")
)
prompt = Prompt.from_function(fn)
assert await prompt.render() == [
AssistantMessage(content=TextContent(type="text", text="Hello, world!"))
]
async def test_fn_returns_multiple_messages(self):
expected = [
UserMessage("Hello, world!"),
AssistantMessage("How can I help you today?"),
UserMessage("I'm looking for a restaurant in the center of town."),
]
async def fn() -> list[Message]:
return expected
prompt = Prompt.from_function(fn)
assert await prompt.render() == expected
async def test_fn_returns_list_of_strings(self):
expected = [
"Hello, world!",
"I'm looking for a restaurant in the center of town.",
]
async def fn() -> list[str]:
return expected
prompt = Prompt.from_function(fn)
assert await prompt.render() == [UserMessage(t) for t in expected]
async def test_fn_returns_resource_content(self):
"""Test returning a message with resource content."""
async def fn() -> UserMessage:
return UserMessage(
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
text="File contents",
mimeType="text/plain",
),
)
)
prompt = Prompt.from_function(fn)
assert await prompt.render() == [
UserMessage(
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
text="File contents",
mimeType="text/plain",
),
)
)
]
async def test_fn_returns_mixed_content(self):
"""Test returning messages with mixed content types."""
async def fn() -> list[Message]:
return [
UserMessage(content="Please analyze this file:"),
UserMessage(
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
text="File contents",
mimeType="text/plain",
),
)
),
AssistantMessage(content="I'll help analyze that file."),
]
prompt = Prompt.from_function(fn)
assert await prompt.render() == [
UserMessage(
content=TextContent(type="text", text="Please analyze this file:")
),
UserMessage(
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
text="File contents",
mimeType="text/plain",
),
)
),
AssistantMessage(
content=TextContent(type="text", text="I'll help analyze that file.")
),
]
async def test_fn_returns_dict_with_resource(self):
"""Test returning a dict with resource content."""
async def fn() -> dict:
return {
"role": "user",
"content": {
"type": "resource",
"resource": {
"uri": FileUrl("file://file.txt"),
"text": "File contents",
"mimeType": "text/plain",
},
},
}
prompt = Prompt.from_function(fn)
assert await prompt.render() == [
UserMessage(
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
text="File contents",
mimeType="text/plain",
),
)
)
]

107
tests/prompts/test_manager.py
View file

@ -1,107 +0,0 @@
import pytest
from fastmcp.prompts.base import UserMessage, TextContent, Prompt
from fastmcp.prompts.manager import PromptManager
class TestPromptManager:
def test_add_prompt(self):
"""Test adding a prompt to the manager."""
def fn() -> str:
return "Hello, world!"
manager = PromptManager()
prompt = Prompt.from_function(fn)
added = manager.add_prompt(prompt)
assert added == prompt
assert manager.get_prompt("fn") == prompt
def test_add_duplicate_prompt(self, caplog):
"""Test adding the same prompt twice."""
def fn() -> str:
return "Hello, world!"
manager = PromptManager()
prompt = Prompt.from_function(fn)
first = manager.add_prompt(prompt)
second = manager.add_prompt(prompt)
assert first == second
assert "Prompt already exists" in caplog.text
def test_disable_warn_on_duplicate_prompts(self, caplog):
"""Test disabling warning on duplicate prompts."""
def fn() -> str:
return "Hello, world!"
manager = PromptManager(warn_on_duplicate_prompts=False)
prompt = Prompt.from_function(fn)
first = manager.add_prompt(prompt)
second = manager.add_prompt(prompt)
assert first == second
assert "Prompt already exists" not in caplog.text
def test_list_prompts(self):
"""Test listing all prompts."""
def fn1() -> str:
return "Hello, world!"
def fn2() -> str:
return "Goodbye, world!"
manager = PromptManager()
prompt1 = Prompt.from_function(fn1)
prompt2 = Prompt.from_function(fn2)
manager.add_prompt(prompt1)
manager.add_prompt(prompt2)
prompts = manager.list_prompts()
assert len(prompts) == 2
assert prompts == [prompt1, prompt2]
async def test_render_prompt(self):
"""Test rendering a prompt."""
def fn() -> str:
return "Hello, world!"
manager = PromptManager()
prompt = Prompt.from_function(fn)
manager.add_prompt(prompt)
messages = await manager.render_prompt("fn")
assert messages == [
UserMessage(content=TextContent(type="text", text="Hello, world!"))
]
async def test_render_prompt_with_args(self):
"""Test rendering a prompt with arguments."""
def fn(name: str) -> str:
return f"Hello, {name}!"
manager = PromptManager()
prompt = Prompt.from_function(fn)
manager.add_prompt(prompt)
messages = await manager.render_prompt("fn", arguments={"name": "World"})
assert messages == [
UserMessage(content=TextContent(type="text", text="Hello, World!"))
]
async def test_render_unknown_prompt(self):
"""Test rendering a non-existent prompt."""
manager = PromptManager()
with pytest.raises(ValueError, match="Unknown prompt: unknown"):
await manager.render_prompt("unknown")
async def test_render_prompt_with_missing_args(self):
"""Test rendering a prompt with missing required arguments."""
def fn(name: str) -> str:
return f"Hello, {name}!"
manager = PromptManager()
prompt = Prompt.from_function(fn)
manager.add_prompt(prompt)
with pytest.raises(ValueError, match="Missing required arguments"):
await manager.render_prompt("fn")

0
tests/resources/__init__.py
View file

115
tests/resources/test_file_resources.py
View file

@ -1,115 +0,0 @@
import os
import pytest
from pathlib import Path
from tempfile import NamedTemporaryFile
from pydantic import FileUrl
from fastmcp.resources import FileResource
@pytest.fixture
def temp_file():
"""Create a temporary file for testing.
File is automatically cleaned up after the test if it still exists.
"""
content = "test content"
with NamedTemporaryFile(mode="w", delete=False) as f:
f.write(content)
path = Path(f.name).resolve()
yield path
try:
path.unlink()
except FileNotFoundError:
pass # File was already deleted by the test
class TestFileResource:
"""Test FileResource functionality."""
def test_file_resource_creation(self, temp_file: Path):
"""Test creating a FileResource."""
resource = FileResource(
uri=FileUrl(temp_file.as_uri()),
name="test",
description="test file",
path=temp_file,
)
assert str(resource.uri) == temp_file.as_uri()
assert resource.name == "test"
assert resource.description == "test file"
assert resource.mime_type == "text/plain" # default
assert resource.path == temp_file
assert resource.is_binary is False # default
def test_file_resource_str_path_conversion(self, temp_file: Path):
"""Test FileResource handles string paths."""
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=Path(str(temp_file)),
)
assert isinstance(resource.path, Path)
assert resource.path.is_absolute()
async def test_read_text_file(self, temp_file: Path):
"""Test reading a text file."""
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
content = await resource.read()
assert content == "test content"
assert resource.mime_type == "text/plain"
async def test_read_binary_file(self, temp_file: Path):
"""Test reading a file as binary."""
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
is_binary=True,
)
content = await resource.read()
assert isinstance(content, bytes)
assert content == b"test content"
def test_relative_path_error(self):
"""Test error on relative path."""
with pytest.raises(ValueError, match="Path must be absolute"):
FileResource(
uri=FileUrl("file:///test.txt"),
name="test",
path=Path("test.txt"),
)
async def test_missing_file_error(self, temp_file: Path):
"""Test error when file doesn't exist."""
# Create path to non-existent file
missing = temp_file.parent / "missing.txt"
resource = FileResource(
uri=FileUrl("file:///missing.txt"),
name="test",
path=missing,
)
with pytest.raises(ValueError, match="Error reading file"):
await resource.read()
@pytest.mark.skipif(
os.name == "nt", reason="File permissions behave differently on Windows"
)
async def test_permission_error(self, temp_file: Path):
"""Test reading a file without permissions."""
temp_file.chmod(0o000) # Remove all permissions
try:
resource = FileResource(
uri=FileUrl(temp_file.as_uri()),
name="test",
path=temp_file,
)
with pytest.raises(ValueError, match="Error reading file"):
await resource.read()
finally:
temp_file.chmod(0o644) # Restore permissions

115
tests/resources/test_function_resources.py
View file

@ -1,115 +0,0 @@
from pydantic import BaseModel, AnyUrl
import pytest
from fastmcp.resources import FunctionResource
class TestFunctionResource:
"""Test FunctionResource functionality."""
def test_function_resource_creation(self):
"""Test creating a FunctionResource."""
def my_func() -> str:
return "test content"
resource = FunctionResource(
uri=AnyUrl("fn://test"),
name="test",
description="test function",
fn=my_func,
)
assert str(resource.uri) == "fn://test"
assert resource.name == "test"
assert resource.description == "test function"
assert resource.mime_type == "text/plain" # default
assert resource.fn == my_func
async def test_read_text(self):
"""Test reading text from a FunctionResource."""
def get_data() -> str:
return "Hello, world!"
resource = FunctionResource(
uri=AnyUrl("function://test"),
name="test",
fn=get_data,
)
content = await resource.read()
assert content == "Hello, world!"
assert resource.mime_type == "text/plain"
async def test_read_binary(self):
"""Test reading binary data from a FunctionResource."""
def get_data() -> bytes:
return b"Hello, world!"
resource = FunctionResource(
uri=AnyUrl("function://test"),
name="test",
fn=get_data,
)
content = await resource.read()
assert content == b"Hello, world!"
async def test_json_conversion(self):
"""Test automatic JSON conversion of non-string results."""
def get_data() -> dict:
return {"key": "value"}
resource = FunctionResource(
uri=AnyUrl("function://test"),
name="test",
fn=get_data,
)
content = await resource.read()
assert isinstance(content, str)
assert '"key": "value"' in content
async def test_error_handling(self):
"""Test error handling in FunctionResource."""
def failing_func() -> str:
raise ValueError("Test error")
resource = FunctionResource(
uri=AnyUrl("function://test"),
name="test",
fn=failing_func,
)
with pytest.raises(ValueError, match="Error reading resource function://test"):
await resource.read()
async def test_basemodel_conversion(self):
"""Test handling of BaseModel types."""
class MyModel(BaseModel):
name: str
resource = FunctionResource(
uri=AnyUrl("function://test"),
name="test",
fn=lambda: MyModel(name="test"),
)
content = await resource.read()
assert content == '{"name": "test"}'
async def test_custom_type_conversion(self):
"""Test handling of custom types."""
class CustomData:
def __str__(self) -> str:
return "custom data"
def get_data() -> CustomData:
return CustomData()
resource = FunctionResource(
uri=AnyUrl("function://test"),
name="test",
fn=get_data,
)
content = await resource.read()
assert isinstance(content, str)

137
tests/resources/test_resource_manager.py
View file

@ -1,137 +0,0 @@
import pytest
from pathlib import Path
from tempfile import NamedTemporaryFile
from pydantic import AnyUrl, FileUrl
from fastmcp.resources import (
FileResource,
FunctionResource,
ResourceManager,
ResourceTemplate,
)
@pytest.fixture
def temp_file():
"""Create a temporary file for testing.
File is automatically cleaned up after the test if it still exists.
"""
content = "test content"
with NamedTemporaryFile(mode="w", delete=False) as f:
f.write(content)
path = Path(f.name).resolve()
yield path
try:
path.unlink()
except FileNotFoundError:
pass # File was already deleted by the test
class TestResourceManager:
"""Test ResourceManager functionality."""
def test_add_resource(self, temp_file: Path):
"""Test adding a resource."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
added = manager.add_resource(resource)
assert added == resource
assert manager.list_resources() == [resource]
def test_add_duplicate_resource(self, temp_file: Path):
"""Test adding the same resource twice."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
first = manager.add_resource(resource)
second = manager.add_resource(resource)
assert first == second
assert manager.list_resources() == [resource]
def test_warn_on_duplicate_resources(self, temp_file: Path, caplog):
"""Test warning on duplicate resources."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
manager.add_resource(resource)
manager.add_resource(resource)
assert "Resource already exists" in caplog.text
def test_disable_warn_on_duplicate_resources(self, temp_file: Path, caplog):
"""Test disabling warning on duplicate resources."""
manager = ResourceManager(warn_on_duplicate_resources=False)
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
manager.add_resource(resource)
manager.add_resource(resource)
assert "Resource already exists" not in caplog.text
async def test_get_resource(self, temp_file: Path):
"""Test getting a resource by URI."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test",
path=temp_file,
)
manager.add_resource(resource)
retrieved = await manager.get_resource(resource.uri)
assert retrieved == resource
async def test_get_resource_from_template(self):
"""Test getting a resource through a template."""
manager = ResourceManager()
def greet(name: str) -> str:
return f"Hello, {name}!"
template = ResourceTemplate.from_function(
fn=greet,
uri_template="greet://{name}",
name="greeter",
)
manager._templates[template.uri_template] = template
resource = await manager.get_resource(AnyUrl("greet://world"))
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "Hello, world!"
async def test_get_unknown_resource(self):
"""Test getting a non-existent resource."""
manager = ResourceManager()
with pytest.raises(ValueError, match="Unknown resource"):
await manager.get_resource(AnyUrl("unknown://test"))
def test_list_resources(self, temp_file: Path):
"""Test listing all resources."""
manager = ResourceManager()
resource1 = FileResource(
uri=FileUrl(f"file://{temp_file}"),
name="test1",
path=temp_file,
)
resource2 = FileResource(
uri=FileUrl(f"file://{temp_file}2"),
name="test2",
path=temp_file,
)
manager.add_resource(resource1)
manager.add_resource(resource2)
resources = manager.list_resources()
assert len(resources) == 2
assert resources == [resource1, resource2]

181
tests/resources/test_resource_template.py
View file

@ -1,181 +0,0 @@
import json
import pytest
from pydantic import BaseModel
from fastmcp.resources import FunctionResource, ResourceTemplate
class TestResourceTemplate:
"""Test ResourceTemplate functionality."""
def test_template_creation(self):
"""Test creating a template from a function."""
def my_func(key: str, value: int) -> dict:
return {"key": key, "value": value}
template = ResourceTemplate.from_function(
fn=my_func,
uri_template="test://{key}/{value}",
name="test",
)
assert template.uri_template == "test://{key}/{value}"
assert template.name == "test"
assert template.mime_type == "text/plain" # default
test_input = {"key": "test", "value": 42}
assert template.fn(**test_input) == my_func(**test_input)
def test_template_matches(self):
"""Test matching URIs against a template."""
def my_func(key: str, value: int) -> dict:
return {"key": key, "value": value}
template = ResourceTemplate.from_function(
fn=my_func,
uri_template="test://{key}/{value}",
name="test",
)
# Valid match
params = template.matches("test://foo/123")
assert params == {"key": "foo", "value": "123"}
# No match
assert template.matches("test://foo") is None
assert template.matches("other://foo/123") is None
async def test_create_resource(self):
"""Test creating a resource from a template."""
def my_func(key: str, value: int) -> dict:
return {"key": key, "value": value}
template = ResourceTemplate.from_function(
fn=my_func,
uri_template="test://{key}/{value}",
name="test",
)
resource = await template.create_resource(
"test://foo/123",
{"key": "foo", "value": 123},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert isinstance(content, str)
data = json.loads(content)
assert data == {"key": "foo", "value": 123}
async def test_template_error(self):
"""Test error handling in template resource creation."""
def failing_func(x: str) -> str:
raise ValueError("Test error")
template = ResourceTemplate.from_function(
fn=failing_func,
uri_template="fail://{x}",
name="fail",
)
with pytest.raises(ValueError, match="Error creating resource from template"):
await template.create_resource("fail://test", {"x": "test"})
async def test_async_text_resource(self):
"""Test creating a text resource from async function."""
async def greet(name: str) -> str:
return f"Hello, {name}!"
template = ResourceTemplate.from_function(
fn=greet,
uri_template="greet://{name}",
name="greeter",
)
resource = await template.create_resource(
"greet://world",
{"name": "world"},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "Hello, world!"
async def test_async_binary_resource(self):
"""Test creating a binary resource from async function."""
async def get_bytes(value: str) -> bytes:
return value.encode()
template = ResourceTemplate.from_function(
fn=get_bytes,
uri_template="bytes://{value}",
name="bytes",
)
resource = await template.create_resource(
"bytes://test",
{"value": "test"},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == b"test"
async def test_basemodel_conversion(self):
"""Test handling of BaseModel types."""
class MyModel(BaseModel):
key: str
value: int
def get_data(key: str, value: int) -> MyModel:
return MyModel(key=key, value=value)
template = ResourceTemplate.from_function(
fn=get_data,
uri_template="test://{key}/{value}",
name="test",
)
resource = await template.create_resource(
"test://foo/123",
{"key": "foo", "value": 123},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert isinstance(content, str)
data = json.loads(content)
assert data == {"key": "foo", "value": 123}
async def test_custom_type_conversion(self):
"""Test handling of custom types."""
class CustomData:
def __init__(self, value: str):
self.value = value
def __str__(self) -> str:
return self.value
def get_data(value: str) -> CustomData:
return CustomData(value)
template = ResourceTemplate.from_function(
fn=get_data,
uri_template="test://{value}",
name="test",
)
resource = await template.create_resource(
"test://hello",
{"value": "hello"},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "hello"

100
tests/resources/test_resources.py
View file

@ -1,100 +0,0 @@
import pytest
from pydantic import AnyUrl
from fastmcp.resources import FunctionResource, Resource
class TestResourceValidation:
"""Test base Resource validation."""
def test_resource_uri_validation(self):
"""Test URI validation."""
def dummy_func() -> str:
return "data"
# Valid URI
resource = FunctionResource(
uri=AnyUrl("http://example.com/data"),
name="test",
fn=dummy_func,
)
assert str(resource.uri) == "http://example.com/data"
# Missing protocol
with pytest.raises(ValueError, match="Input should be a valid URL"):
FunctionResource(
uri=AnyUrl("invalid"),
name="test",
fn=dummy_func,
)
# Missing host
with pytest.raises(ValueError, match="Input should be a valid URL"):
FunctionResource(
uri=AnyUrl("http://"),
name="test",
fn=dummy_func,
)
def test_resource_name_from_uri(self):
"""Test name is extracted from URI if not provided."""
def dummy_func() -> str:
return "data"
resource = FunctionResource(
uri=AnyUrl("resource://my-resource"),
fn=dummy_func,
)
assert resource.name == "resource://my-resource"
def test_resource_name_validation(self):
"""Test name validation."""
def dummy_func() -> str:
return "data"
# Must provide either name or URI
with pytest.raises(ValueError, match="Either name or uri must be provided"):
FunctionResource(
fn=dummy_func,
)
# Explicit name takes precedence over URI
resource = FunctionResource(
uri=AnyUrl("resource://uri-name"),
name="explicit-name",
fn=dummy_func,
)
assert resource.name == "explicit-name"
def test_resource_mime_type(self):
"""Test mime type handling."""
def dummy_func() -> str:
return "data"
# Default mime type
resource = FunctionResource(
uri=AnyUrl("resource://test"),
fn=dummy_func,
)
assert resource.mime_type == "text/plain"
# Custom mime type
resource = FunctionResource(
uri=AnyUrl("resource://test"),
fn=dummy_func,
mime_type="application/json",
)
assert resource.mime_type == "application/json"
async def test_resource_read_abstract(self):
"""Test that Resource.read() is abstract."""
class ConcreteResource(Resource):
pass
with pytest.raises(TypeError, match="abstract method"):
ConcreteResource(uri=AnyUrl("test://test"), name="test") # type: ignore

0
tests/servers/__init__.py
View file

114
tests/servers/test_file_server.py
View file

@ -1,114 +0,0 @@
import json
from fastmcp import FastMCP
import pytest
from pathlib import Path
@pytest.fixture()
def test_dir(tmp_path_factory) -> Path:
"""Create a temporary directory with test files."""
tmp = tmp_path_factory.mktemp("test_files")
# Create test files
(tmp / "example.py").write_text("print('hello world')")
(tmp / "readme.md").write_text("# Test Directory\nThis is a test.")
(tmp / "config.json").write_text('{"test": true}')
return tmp
@pytest.fixture
def mcp() -> FastMCP:
mcp = FastMCP()
return mcp
@pytest.fixture(autouse=True)
def resources(mcp: FastMCP, test_dir: Path) -> FastMCP:
@mcp.resource("dir://test_dir")
def list_test_dir() -> list[str]:
"""List the files in the test directory"""
return [str(f) for f in test_dir.iterdir()]
@mcp.resource("file://test_dir/example.py")
def read_example_py() -> str:
"""Read the example.py file"""
try:
return (test_dir / "example.py").read_text()
except FileNotFoundError:
return "File not found"
@mcp.resource("file://test_dir/readme.md")
def read_readme_md() -> str:
"""Read the readme.md file"""
try:
return (test_dir / "readme.md").read_text()
except FileNotFoundError:
return "File not found"
@mcp.resource("file://test_dir/config.json")
def read_config_json() -> str:
"""Read the config.json file"""
try:
return (test_dir / "config.json").read_text()
except FileNotFoundError:
return "File not found"
return mcp
@pytest.fixture(autouse=True)
def tools(mcp: FastMCP, test_dir: Path) -> FastMCP:
@mcp.tool()
def delete_file(path: str) -> bool:
# ensure path is in test_dir
if Path(path).resolve().parent != test_dir:
raise ValueError(f"Path must be in test_dir: {path}")
Path(path).unlink()
return True
return mcp
async def test_list_resources(mcp: FastMCP):
resources = await mcp.list_resources()
assert len(resources) == 4
assert [str(r.uri) for r in resources] == [
"dir://test_dir",
"file://test_dir/example.py",
"file://test_dir/readme.md",
"file://test_dir/config.json",
]
async def test_read_resource_dir(mcp: FastMCP):
files = await mcp.read_resource("dir://test_dir")
files = json.loads(files)
assert sorted([Path(f).name for f in files]) == [
"config.json",
"example.py",
"readme.md",
]
async def test_read_resource_file(mcp: FastMCP):
result = await mcp.read_resource("file://test_dir/example.py")
assert result == "print('hello world')"
async def test_delete_file(mcp: FastMCP, test_dir: Path):
await mcp.call_tool(
"delete_file", arguments=dict(path=str(test_dir / "example.py"))
)
assert not (test_dir / "example.py").exists()
async def test_delete_file_and_check_resources(mcp: FastMCP, test_dir: Path):
await mcp.call_tool(
"delete_file", arguments=dict(path=str(test_dir / "example.py"))
)
result = await mcp.read_resource("file://test_dir/example.py")
assert result == "File not found"

376
tests/test_cli.py
View file

@ -1,376 +0,0 @@
"""Tests for the FastMCP CLI."""
import json
import sys
from pathlib import Path
from unittest.mock import call, patch
import pytest
from typer.testing import CliRunner
from fastmcp.cli.cli import _parse_env_var, _parse_file_path, app
@pytest.fixture
def mock_config(tmp_path):
"""Create a mock Claude config file."""
config = {"mcpServers": {}}
config_file = tmp_path / "claude_desktop_config.json"
config_file.write_text(json.dumps(config))
return config_file
@pytest.fixture
def server_file(tmp_path):
"""Create a server file."""
server_file = tmp_path / "server.py"
server_file.write_text(
"""from fastmcp import FastMCP
mcp = FastMCP("test")
"""
)
return server_file
@pytest.fixture
def mock_env_file(tmp_path):
"""Create a mock .env file."""
env_file = tmp_path / ".env"
env_file.write_text("FOO=bar\nBAZ=123")
return env_file
def test_parse_env_var():
"""Test parsing environment variables."""
assert _parse_env_var("FOO=bar") == ("FOO", "bar")
assert _parse_env_var("FOO=") == ("FOO", "")
assert _parse_env_var("FOO=bar baz") == ("FOO", "bar baz")
assert _parse_env_var("FOO = bar ") == ("FOO", "bar")
with pytest.raises(SystemExit):
_parse_env_var("invalid")
@pytest.mark.parametrize(
"args,expected_env",
[
# Basic env var
(
["--env-var", "FOO=bar"],
{"FOO": "bar"},
),
# Multiple env vars
(
["--env-var", "FOO=bar", "--env-var", "BAZ=123"],
{"FOO": "bar", "BAZ": "123"},
),
# Env var with spaces
(
["--env-var", "FOO=bar baz"],
{"FOO": "bar baz"},
),
],
)
def test_install_with_env_vars(mock_config, server_file, args, expected_env):
"""Test installing with environment variables."""
runner = CliRunner()
with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
mock_config_path.return_value = mock_config.parent
result = runner.invoke(
app,
["install", str(server_file)] + args,
)
assert result.exit_code == 0
# Read the config file and check env vars
config = json.loads(mock_config.read_text())
assert "mcpServers" in config
assert len(config["mcpServers"]) == 1
server = next(iter(config["mcpServers"].values()))
assert server["env"] == expected_env
def test_parse_file_path_windows_drive():
"""Test parsing a Windows file path with a drive letter."""
file_spec = r"C:\path\to\file.txt"
with (
patch("pathlib.Path.exists", return_value=True),
patch("pathlib.Path.is_file", return_value=True),
):
file_path, server_object = _parse_file_path(file_spec)
assert file_path == Path(r"C:\path\to\file.txt").resolve()
assert server_object is None
def test_parse_file_path_with_object():
"""Test parsing a file path with an object specification."""
file_spec = "/path/to/file.txt:object"
with patch("sys.exit") as mock_exit:
_parse_file_path(file_spec)
# Check that sys.exit was called twice with code 1
assert mock_exit.call_count == 2
mock_exit.assert_has_calls([call(1), call(1)])
def test_parse_file_path_windows_with_object():
"""Test parsing a Windows file path with an object specification."""
file_spec = r"C:\path\to\file.txt:object"
with (
patch("pathlib.Path.exists", return_value=True),
patch("pathlib.Path.is_file", return_value=True),
):
file_path, server_object = _parse_file_path(file_spec)
assert file_path == Path(r"C:\path\to\file.txt").resolve()
assert server_object == "object"
def test_install_with_env_file(mock_config, server_file, mock_env_file):
"""Test installing with environment variables from a file."""
runner = CliRunner()
with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
mock_config_path.return_value = mock_config.parent
result = runner.invoke(
app,
["install", str(server_file), "--env-file", str(mock_env_file)],
)
assert result.exit_code == 0
# Read the config file and check env vars
config = json.loads(mock_config.read_text())
assert "mcpServers" in config
assert len(config["mcpServers"]) == 1
server = next(iter(config["mcpServers"].values()))
assert server["env"] == {"FOO": "bar", "BAZ": "123"}
def test_install_preserves_existing_env_vars(mock_config, server_file):
"""Test that installing preserves existing environment variables."""
# Set up initial config with env vars
config = {
"mcpServers": {
"test": {
"command": "uv",
"args": [
"run",
"--with",
"fastmcp",
"fastmcp",
"run",
str(server_file),
],
"env": {"FOO": "bar", "BAZ": "123"},
}
}
}
mock_config.write_text(json.dumps(config))
runner = CliRunner()
with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
mock_config_path.return_value = mock_config.parent
# Install with a new env var
result = runner.invoke(
app,
["install", str(server_file), "--env-var", "NEW=value"],
)
assert result.exit_code == 0
# Read the config file and check env vars are preserved
config = json.loads(mock_config.read_text())
server = next(iter(config["mcpServers"].values()))
assert server["env"] == {"FOO": "bar", "BAZ": "123", "NEW": "value"}
def test_install_updates_existing_env_vars(mock_config, server_file):
"""Test that installing updates existing environment variables."""
# Set up initial config with env vars
config = {
"mcpServers": {
"test": {
"command": "uv",
"args": [
"run",
"--with",
"fastmcp",
"fastmcp",
"run",
str(server_file),
],
"env": {"FOO": "bar", "BAZ": "123"},
}
}
}
mock_config.write_text(json.dumps(config))
runner = CliRunner()
with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
mock_config_path.return_value = mock_config.parent
# Update an existing env var
result = runner.invoke(
app,
["install", str(server_file), "--env-var", "FOO=newvalue"],
)
assert result.exit_code == 0
# Read the config file and check env var was updated
config = json.loads(mock_config.read_text())
server = next(iter(config["mcpServers"].values()))
assert server["env"] == {"FOO": "newvalue", "BAZ": "123"}
def test_server_dependencies(mock_config, server_file):
"""Test that server dependencies are correctly handled."""
# Create a server file with dependencies
server_file = server_file.parent / "server_with_deps.py"
server_file.write_text(
"""from fastmcp import FastMCP
mcp = FastMCP("test", dependencies=["pandas", "numpy"])
"""
)
runner = CliRunner()
with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
mock_config_path.return_value = mock_config.parent
result = runner.invoke(app, ["install", str(server_file)])
assert result.exit_code == 0
# Read the config file and check dependencies were added as --with args
config = json.loads(mock_config.read_text())
server = next(iter(config["mcpServers"].values()))
assert "--with" in server["args"]
assert "pandas" in server["args"]
assert "numpy" in server["args"]
def test_server_dependencies_empty(mock_config, server_file):
"""Test that server with no dependencies works correctly."""
runner = CliRunner()
with patch("fastmcp.cli.claude.get_claude_config_path") as mock_config_path:
mock_config_path.return_value = mock_config.parent
result = runner.invoke(app, ["install", str(server_file)])
assert result.exit_code == 0
# Read the config file and check only fastmcp is in --with args
config = json.loads(mock_config.read_text())
server = next(iter(config["mcpServers"].values()))
assert server["args"].count("--with") == 1
assert "fastmcp" in server["args"]
def test_dev_with_dependencies(mock_config, server_file):
"""Test that dev command handles dependencies correctly."""
server_file = server_file.parent / "server_with_deps.py"
server_file.write_text(
"""from fastmcp import FastMCP
mcp = FastMCP("test", dependencies=["pandas", "numpy"])
"""
)
runner = CliRunner()
with patch("subprocess.run") as mock_run:
mock_run.return_value.returncode = 0
result = runner.invoke(app, ["dev", str(server_file)])
assert result.exit_code == 0
if sys.platform == "win32":
# On Windows, expect two calls
assert mock_run.call_count == 2
assert mock_run.call_args_list[0] == call(
["npx.cmd", "--version"], check=True, capture_output=True, shell=True
)
# get the actual command and expected command without dependencies
actual_cmd = mock_run.call_args_list[1][0][0]
expected_start = [
"npx.cmd",
"@modelcontextprotocol/inspector",
"uv",
"run",
"--with",
"fastmcp",
]
expected_end = ["fastmcp", "run", str(server_file)]
# verify start and end of command
assert actual_cmd[: len(expected_start)] == expected_start
assert actual_cmd[-len(expected_end) :] == expected_end
# verify dependencies are present (order-independent)
deps_section = actual_cmd[len(expected_start) : -len(expected_end)]
assert all(
x in deps_section for x in ["--with", "numpy", "--with", "pandas"]
)
# Verify subprocess call kwargs, allowing for environment variables
call_kwargs = mock_run.call_args_list[1][1]
assert call_kwargs["check"] is True
assert call_kwargs["shell"] is True
assert isinstance(call_kwargs["env"], dict)
else:
# same verification for unix, just with different command prefix
actual_cmd = mock_run.call_args_list[0][0][0]
expected_start = [
"npx",
"@modelcontextprotocol/inspector",
"uv",
"run",
"--with",
"fastmcp",
]
expected_end = ["fastmcp", "run", str(server_file)]
assert actual_cmd[: len(expected_start)] == expected_start
assert actual_cmd[-len(expected_end) :] == expected_end
deps_section = actual_cmd[len(expected_start) : -len(expected_end)]
assert all(
x in deps_section for x in ["--with", "numpy", "--with", "pandas"]
)
# Verify subprocess call kwargs, allowing for environment variables
call_kwargs = mock_run.call_args_list[0][1]
assert call_kwargs["check"] is True
assert call_kwargs["shell"] is False
assert isinstance(call_kwargs["env"], dict)
def test_run_with_dependencies(mock_config, server_file):
"""Test that run command does not handle dependencies."""
# Create a server file with dependencies
server_file = server_file.parent / "server_with_deps.py"
server_file.write_text(
"""from fastmcp import FastMCP
mcp = FastMCP("test", dependencies=["pandas", "numpy"])
if __name__ == "__main__":
mcp.run()
"""
)
runner = CliRunner()
with patch("subprocess.run") as mock_run:
result = runner.invoke(app, ["run", str(server_file)])
assert result.exit_code == 0
# Run command should not call subprocess.run
mock_run.assert_not_called()

376
tests/test_func_metadata.py
View file

@ -1,376 +0,0 @@
from typing import Annotated
import annotated_types
import pytest
from pydantic import BaseModel, Field
from fastmcp.utilities.func_metadata import func_metadata
class SomeInputModelA(BaseModel):
pass
class SomeInputModelB(BaseModel):
class InnerModel(BaseModel):
x: int
how_many_shrimp: Annotated[int, Field(description="How many shrimp in the tank???")]
ok: InnerModel
y: None
def complex_arguments_fn(
an_int: int,
must_be_none: None,
must_be_none_dumb_annotation: Annotated[None, "blah"],
list_of_ints: list[int],
# list[str] | str is an interesting case because if it comes in as JSON like
# "[\"a\", \"b\"]" then it will be naively parsed as a string.
list_str_or_str: list[str] | str,
an_int_annotated_with_field: Annotated[
int, Field(description="An int with a field")
],
an_int_annotated_with_field_and_others: Annotated[
int,
str, # Should be ignored, really
Field(description="An int with a field"),
annotated_types.Gt(1),
],
an_int_annotated_with_junk: Annotated[
int,
"123",
456,
],
field_with_default_via_field_annotation_before_nondefault_arg: Annotated[
int, Field(1)
],
unannotated,
my_model_a: SomeInputModelA,
my_model_a_forward_ref: "SomeInputModelA",
my_model_b: SomeInputModelB,
an_int_annotated_with_field_default: Annotated[
int,
Field(1, description="An int with a field"),
],
unannotated_with_default=5,
my_model_a_with_default: SomeInputModelA = SomeInputModelA(), # noqa: B008
an_int_with_default: int = 1,
must_be_none_with_default: None = None,
an_int_with_equals_field: int = Field(1, ge=0),
int_annotated_with_default: Annotated[int, Field(description="hey")] = 5,
) -> str:
_ = (
an_int,
must_be_none,
must_be_none_dumb_annotation,
list_of_ints,
list_str_or_str,
an_int_annotated_with_field,
an_int_annotated_with_field_and_others,
an_int_annotated_with_junk,
field_with_default_via_field_annotation_before_nondefault_arg,
unannotated,
an_int_annotated_with_field_default,
unannotated_with_default,
my_model_a,
my_model_a_forward_ref,
my_model_b,
my_model_a_with_default,
an_int_with_default,
must_be_none_with_default,
an_int_with_equals_field,
int_annotated_with_default,
)
return "ok!"
async def test_complex_function_runtime_arg_validation_non_json():
"""Test that basic non-JSON arguments are validated correctly"""
meta = func_metadata(complex_arguments_fn)
# Test with minimum required arguments
result = await meta.call_fn_with_arg_validation(
complex_arguments_fn,
fn_is_async=False,
arguments_to_validate={
"an_int": 1,
"must_be_none": None,
"must_be_none_dumb_annotation": None,
"list_of_ints": [1, 2, 3],
"list_str_or_str": "hello",
"an_int_annotated_with_field": 42,
"an_int_annotated_with_field_and_others": 5,
"an_int_annotated_with_junk": 100,
"unannotated": "test",
"my_model_a": {},
"my_model_a_forward_ref": {},
"my_model_b": {"how_many_shrimp": 5, "ok": {"x": 1}, "y": None},
},
arguments_to_pass_directly=None,
)
assert result == "ok!"
# Test with invalid types
with pytest.raises(ValueError):
await meta.call_fn_with_arg_validation(
complex_arguments_fn,
fn_is_async=False,
arguments_to_validate={"an_int": "not an int"},
arguments_to_pass_directly=None,
)
async def test_complex_function_runtime_arg_validation_with_json():
"""Test that JSON string arguments are parsed and validated correctly"""
meta = func_metadata(complex_arguments_fn)
result = await meta.call_fn_with_arg_validation(
complex_arguments_fn,
fn_is_async=False,
arguments_to_validate={
"an_int": 1,
"must_be_none": None,
"must_be_none_dumb_annotation": None,
"list_of_ints": "[1, 2, 3]", # JSON string
"list_str_or_str": '["a", "b", "c"]', # JSON string
"an_int_annotated_with_field": 42,
"an_int_annotated_with_field_and_others": "5", # JSON string
"an_int_annotated_with_junk": 100,
"unannotated": "test",
"my_model_a": "{}", # JSON string
"my_model_a_forward_ref": "{}", # JSON string
"my_model_b": '{"how_many_shrimp": 5, "ok": {"x": 1}, "y": null}', # JSON string
},
arguments_to_pass_directly=None,
)
assert result == "ok!"
def test_str_vs_list_str():
"""Test handling of string vs list[str] type annotations.
This is tricky as '"hello"' can be parsed as a JSON string or a Python string.
We want to make sure it's kept as a python string.
"""
def func_with_str_types(str_or_list: str | list[str]):
return str_or_list
meta = func_metadata(func_with_str_types)
# Test string input for union type
result = meta.pre_parse_json({"str_or_list": "hello"})
assert result["str_or_list"] == "hello"
# Test string input that contains valid JSON for union type
# We want to see here that the JSON-vali string is NOT parsed as JSON, but rather
# kept as a raw string
result = meta.pre_parse_json({"str_or_list": '"hello"'})
assert result["str_or_list"] == '"hello"'
# Test list input for union type
result = meta.pre_parse_json({"str_or_list": '["hello", "world"]'})
assert result["str_or_list"] == ["hello", "world"]
def test_str_vs_int():
"""
Test that string values are kept as strings even when they contain numbers,
while numbers are parsed correctly.
"""
def func_with_str_and_int(a: str, b: int):
return a
meta = func_metadata(func_with_str_and_int)
result = meta.pre_parse_json({"a": "123", "b": 123})
assert result["a"] == "123"
assert result["b"] == 123
def test_skip_names():
"""Test that skipped parameters are not included in the model"""
def func_with_many_params(
keep_this: int, skip_this: str, also_keep: float, also_skip: bool
):
return keep_this, skip_this, also_keep, also_skip
# Skip some parameters
meta = func_metadata(func_with_many_params, skip_names=["skip_this", "also_skip"])
# Check model fields
assert "keep_this" in meta.arg_model.model_fields
assert "also_keep" in meta.arg_model.model_fields
assert "skip_this" not in meta.arg_model.model_fields
assert "also_skip" not in meta.arg_model.model_fields
# Validate that we can call with only non-skipped parameters
model: BaseModel = meta.arg_model.model_validate({"keep_this": 1, "also_keep": 2.5}) # type: ignore
assert model.keep_this == 1 # type: ignore
assert model.also_keep == 2.5 # type: ignore
async def test_lambda_function():
"""Test lambda function schema and validation"""
fn = lambda x, y=5: x # noqa: E731
meta = func_metadata(lambda x, y=5: x)
# Test schema
assert meta.arg_model.model_json_schema() == {
"properties": {
"x": {"title": "x", "type": "string"},
"y": {"default": 5, "title": "y", "type": "string"},
},
"required": ["x"],
"title": "<lambda>Arguments",
"type": "object",
}
async def check_call(args):
return await meta.call_fn_with_arg_validation(
fn,
fn_is_async=False,
arguments_to_validate=args,
arguments_to_pass_directly=None,
)
# Basic calls
assert await check_call({"x": "hello"}) == "hello"
assert await check_call({"x": "hello", "y": "world"}) == "hello"
assert await check_call({"x": '"hello"'}) == '"hello"'
# Missing required arg
with pytest.raises(ValueError):
await check_call({"y": "world"})
def test_complex_function_json_schema():
meta = func_metadata(complex_arguments_fn)
assert meta.arg_model.model_json_schema() == {
"$defs": {
"InnerModel": {
"properties": {"x": {"title": "X", "type": "integer"}},
"required": ["x"],
"title": "InnerModel",
"type": "object",
},
"SomeInputModelA": {
"properties": {},
"title": "SomeInputModelA",
"type": "object",
},
"SomeInputModelB": {
"properties": {
"how_many_shrimp": {
"description": "How many shrimp in the tank???",
"title": "How Many Shrimp",
"type": "integer",
},
"ok": {"$ref": "#/$defs/InnerModel"},
"y": {"title": "Y", "type": "null"},
},
"required": ["how_many_shrimp", "ok", "y"],
"title": "SomeInputModelB",
"type": "object",
},
},
"properties": {
"an_int": {"title": "An Int", "type": "integer"},
"must_be_none": {"title": "Must Be None", "type": "null"},
"must_be_none_dumb_annotation": {
"title": "Must Be None Dumb Annotation",
"type": "null",
},
"list_of_ints": {
"items": {"type": "integer"},
"title": "List Of Ints",
"type": "array",
},
"list_str_or_str": {
"anyOf": [
{"items": {"type": "string"}, "type": "array"},
{"type": "string"},
],
"title": "List Str Or Str",
},
"an_int_annotated_with_field": {
"description": "An int with a field",
"title": "An Int Annotated With Field",
"type": "integer",
},
"an_int_annotated_with_field_and_others": {
"description": "An int with a field",
"exclusiveMinimum": 1,
"title": "An Int Annotated With Field And Others",
"type": "integer",
},
"an_int_annotated_with_junk": {
"title": "An Int Annotated With Junk",
"type": "integer",
},
"field_with_default_via_field_annotation_before_nondefault_arg": {
"default": 1,
"title": "Field With Default Via Field Annotation Before Nondefault Arg",
"type": "integer",
},
"unannotated": {"title": "unannotated", "type": "string"},
"my_model_a": {"$ref": "#/$defs/SomeInputModelA"},
"my_model_a_forward_ref": {"$ref": "#/$defs/SomeInputModelA"},
"my_model_b": {"$ref": "#/$defs/SomeInputModelB"},
"an_int_annotated_with_field_default": {
"default": 1,
"description": "An int with a field",
"title": "An Int Annotated With Field Default",
"type": "integer",
},
"unannotated_with_default": {
"default": 5,
"title": "unannotated_with_default",
"type": "string",
},
"my_model_a_with_default": {
"$ref": "#/$defs/SomeInputModelA",
"default": {},
},
"an_int_with_default": {
"default": 1,
"title": "An Int With Default",
"type": "integer",
},
"must_be_none_with_default": {
"default": None,
"title": "Must Be None With Default",
"type": "null",
},
"an_int_with_equals_field": {
"default": 1,
"minimum": 0,
"title": "An Int With Equals Field",
"type": "integer",
},
"int_annotated_with_default": {
"default": 5,
"description": "hey",
"title": "Int Annotated With Default",
"type": "integer",
},
},
"required": [
"an_int",
"must_be_none",
"must_be_none_dumb_annotation",
"list_of_ints",
"list_str_or_str",
"an_int_annotated_with_field",
"an_int_annotated_with_field_and_others",
"an_int_annotated_with_junk",
"unannotated",
"my_model_a",
"my_model_a_forward_ref",
"my_model_b",
],
"title": "complex_arguments_fnArguments",
"type": "object",
}

656
tests/test_server.py
View file

@ -1,656 +0,0 @@
import base64
from pathlib import Path
from typing import TYPE_CHECKING, Union
import pytest
from mcp.shared.exceptions import McpError
from mcp.shared.memory import (
create_connected_server_and_client_session as client_session,
)
from mcp.types import (
ImageContent,
TextContent,
TextResourceContents,
BlobResourceContents,
)
from pydantic import AnyUrl
from fastmcp import Context, FastMCP
from fastmcp.prompts.base import EmbeddedResource, Message, UserMessage
from fastmcp.resources import FileResource, FunctionResource
from fastmcp.utilities.types import Image
if TYPE_CHECKING:
from fastmcp import Context
class TestServer:
async def test_create_server(self):
mcp = FastMCP()
assert mcp.name == "FastMCP"
async def test_add_tool_decorator(self):
mcp = FastMCP()
@mcp.tool()
def add(x: int, y: int) -> int:
return x + y
assert len(mcp._tool_manager.list_tools()) == 1
async def test_add_tool_decorator_incorrect_usage(self):
mcp = FastMCP()
with pytest.raises(TypeError, match="The @tool decorator was used incorrectly"):
@mcp.tool # Missing parentheses #type: ignore
def add(x: int, y: int) -> int:
return x + y
async def test_add_resource_decorator(self):
mcp = FastMCP()
@mcp.resource("r://{x}")
def get_data(x: str) -> str:
return f"Data: {x}"
assert len(mcp._resource_manager._templates) == 1
async def test_add_resource_decorator_incorrect_usage(self):
mcp = FastMCP()
with pytest.raises(
TypeError, match="The @resource decorator was used incorrectly"
):
@mcp.resource # Missing parentheses #type: ignore
def get_data(x: str) -> str:
return f"Data: {x}"
def tool_fn(x: int, y: int) -> int:
return x + y
def error_tool_fn() -> None:
raise ValueError("Test error")
def image_tool_fn(path: str) -> Image:
return Image(path)
def mixed_content_tool_fn() -> list[Union[TextContent, ImageContent]]:
return [
TextContent(type="text", text="Hello"),
ImageContent(type="image", data="abc", mimeType="image/png"),
]
class TestServerTools:
async def test_add_tool(self):
mcp = FastMCP()
mcp.add_tool(tool_fn)
mcp.add_tool(tool_fn)
assert len(mcp._tool_manager.list_tools()) == 1
async def test_list_tools(self):
mcp = FastMCP()
mcp.add_tool(tool_fn)
async with client_session(mcp._mcp_server) as client:
tools = await client.list_tools()
assert len(tools.tools) == 1
async def test_call_tool(self):
mcp = FastMCP()
mcp.add_tool(tool_fn)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("my_tool", {"arg1": "value"})
assert not hasattr(result, "error")
assert len(result.content) > 0
async def test_tool_exception_handling(self):
mcp = FastMCP()
mcp.add_tool(error_tool_fn)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("error_tool_fn", {})
assert len(result.content) == 1
content = result.content[0]
assert isinstance(content, TextContent)
assert "Test error" in content.text
assert result.isError is True
async def test_tool_error_handling(self):
mcp = FastMCP()
mcp.add_tool(error_tool_fn)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("error_tool_fn", {})
assert len(result.content) == 1
content = result.content[0]
assert isinstance(content, TextContent)
assert "Test error" in content.text
assert result.isError is True
async def test_tool_error_details(self):
"""Test that exception details are properly formatted in the response"""
mcp = FastMCP()
mcp.add_tool(error_tool_fn)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("error_tool_fn", {})
content = result.content[0]
assert isinstance(content, TextContent)
assert isinstance(content.text, str)
assert "Test error" in content.text
assert result.isError is True
async def test_tool_return_value_conversion(self):
mcp = FastMCP()
mcp.add_tool(tool_fn)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("tool_fn", {"x": 1, "y": 2})
assert len(result.content) == 1
content = result.content[0]
assert isinstance(content, TextContent)
assert content.text == "3"
async def test_tool_image_helper(self, tmp_path: Path):
# Create a test image
image_path = tmp_path / "test.png"
image_path.write_bytes(b"fake png data")
mcp = FastMCP()
mcp.add_tool(image_tool_fn)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("image_tool_fn", {"path": str(image_path)})
assert len(result.content) == 1
content = result.content[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):
mcp = FastMCP()
mcp.add_tool(mixed_content_tool_fn)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("mixed_content_tool_fn", {})
assert len(result.content) == 2
content1 = result.content[0]
content2 = result.content[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, tmp_path: Path):
"""Test that lists containing Image objects and other types are handled correctly"""
# Create a test image
image_path = tmp_path / "test.png"
image_path.write_bytes(b"test image data")
def mixed_list_fn() -> list:
return [
"text message",
Image(image_path),
{"key": "value"},
TextContent(type="text", text="direct content"),
]
mcp = FastMCP()
mcp.add_tool(mixed_list_fn)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("mixed_list_fn", {})
assert len(result.content) == 4
# Check text conversion
content1 = result.content[0]
assert isinstance(content1, TextContent)
assert content1.text == "text message"
# Check image conversion
content2 = result.content[1]
assert isinstance(content2, ImageContent)
assert content2.mimeType == "image/png"
assert base64.b64decode(content2.data) == b"test image data"
# Check dict conversion
content3 = result.content[2]
assert isinstance(content3, TextContent)
assert '"key": "value"' in content3.text
# Check direct TextContent
content4 = result.content[3]
assert isinstance(content4, TextContent)
assert content4.text == "direct content"
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_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://test"))
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[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_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://binary"))
assert isinstance(result.contents[0], BlobResourceContents)
assert result.contents[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_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("file://test.txt"))
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[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_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("file://test.bin"))
assert isinstance(result.contents[0], BlobResourceContents)
assert (
result.contents[0].blob
== base64.b64encode(b"Binary file data").decode()
)
class TestServerResourceTemplates:
async def test_resource_with_params(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="Mismatch between URI parameters"):
@mcp.resource("resource://data")
def get_data_fn(param: str) -> str:
return f"Data: {param}"
async def test_resource_with_uri_params(self):
"""Test that a resource with URI parameters is automatically a template"""
mcp = FastMCP()
with pytest.raises(ValueError, match="Mismatch between URI parameters"):
@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_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://test/data"))
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[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="Mismatch between URI parameters"):
@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_session(mcp._mcp_server) as client:
result = await client.read_resource(
AnyUrl("resource://cursor/fastmcp/data")
)
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[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="Mismatch between URI parameters"):
@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_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://static"))
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].text == "Static data"
async def test_template_to_resource_conversion(self):
"""Test that templates are properly converted to resources when accessed"""
mcp = FastMCP()
@mcp.resource("resource://{name}/data")
def get_data(name: str) -> str:
return f"Data for {name}"
# Should be registered as a template
assert len(mcp._resource_manager._templates) == 1
assert len(await mcp.list_resources()) == 0
# 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"
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(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_session(mcp._mcp_server) as client:
result = await client.call_tool("tool_with_context", {"x": 42})
assert len(result.content) == 1
content = result.content[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_session(mcp._mcp_server) as client:
result = await client.call_tool("async_tool", {"x": 42})
assert len(result.content) == 1
content = result.content[0]
assert isinstance(content, TextContent)
assert "Async request" in content.text
assert "42" in content.text
async def test_context_logging(self):
"""Test that context logging methods work."""
mcp = FastMCP()
def logging_tool(msg: str, ctx: Context) -> str:
ctx.debug("Debug message")
ctx.info("Info message")
ctx.warning("Warning message")
ctx.error("Error message")
return f"Logged messages for {msg}"
mcp.add_tool(logging_tool)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("logging_tool", {"msg": "test"})
assert len(result.content) == 1
content = result.content[0]
assert isinstance(content, TextContent)
assert "Logged messages for test" in content.text
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_session(mcp._mcp_server) as client:
result = await client.call_tool("no_context", {"x": 21})
assert len(result.content) == 1
content = result.content[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:
data = await ctx.read_resource("test://data")
return f"Read resource: {data}"
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("tool_with_resource", {})
assert len(result.content) == 1
content = result.content[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 = mcp._prompt_manager.list_prompts()
assert len(prompts) == 1
assert prompts[0].name == "fn"
# Don't compare functions directly since validate_call wraps them
content = await prompts[0].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 = mcp._prompt_manager.list_prompts()
assert len(prompts) == 1
assert prompts[0].name == "custom_name"
content = await prompts[0].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 = mcp._prompt_manager.list_prompts()
assert len(prompts) == 1
assert prompts[0].description == "A custom description"
content = await prompts[0].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}!"
async with client_session(mcp._mcp_server) as client:
result = await client.list_prompts()
assert result.prompts is not None
assert len(result.prompts) == 1
prompt = result.prompts[0]
assert prompt.name == "fn"
assert prompt.arguments is not None
assert len(prompt.arguments) == 2
assert prompt.arguments[0].name == "name"
assert prompt.arguments[0].required is True
assert prompt.arguments[1].name == "optional"
assert prompt.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_session(mcp._mcp_server) as client:
result = await client.get_prompt("fn", {"name": "World"})
assert len(result.messages) == 1
message = result.messages[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_session(mcp._mcp_server) as client:
result = await client.get_prompt("fn")
assert len(result.messages) == 1
message = result.messages[0]
assert message.role == "user"
content = message.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()
async with client_session(mcp._mcp_server) as client:
with pytest.raises(McpError, match="Unknown prompt"):
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}!"
async with client_session(mcp._mcp_server) as client:
with pytest.raises(McpError, match="Missing required arguments"):
await client.get_prompt("prompt_fn")

306
tests/test_tool_manager.py
View file

@ -1,306 +0,0 @@
import logging
from typing import Optional
import pytest
from pydantic import BaseModel
import json
from fastmcp.exceptions import ToolError
from fastmcp.tools import ToolManager
class TestAddTools:
def test_basic_function(self):
"""Test registering and running a basic function."""
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
manager = ToolManager()
manager.add_tool(add)
tool = manager.get_tool("add")
assert tool is not None
assert tool.name == "add"
assert tool.description == "Add two numbers."
assert tool.is_async is False
assert tool.parameters["properties"]["a"]["type"] == "integer"
assert tool.parameters["properties"]["b"]["type"] == "integer"
async def test_async_function(self):
"""Test registering and running an async function."""
async def fetch_data(url: str) -> str:
"""Fetch data from URL."""
return f"Data from {url}"
manager = ToolManager()
manager.add_tool(fetch_data)
tool = manager.get_tool("fetch_data")
assert tool is not None
assert tool.name == "fetch_data"
assert tool.description == "Fetch data from URL."
assert tool.is_async is True
assert tool.parameters["properties"]["url"]["type"] == "string"
def test_pydantic_model_function(self):
"""Test registering a function that takes a Pydantic model."""
class UserInput(BaseModel):
name: str
age: int
def create_user(user: UserInput, flag: bool) -> dict:
"""Create a new user."""
return {"id": 1, **user.model_dump()}
manager = ToolManager()
manager.add_tool(create_user)
tool = manager.get_tool("create_user")
assert tool is not None
assert tool.name == "create_user"
assert tool.description == "Create a new user."
assert tool.is_async is False
assert "name" in tool.parameters["$defs"]["UserInput"]["properties"]
assert "age" in tool.parameters["$defs"]["UserInput"]["properties"]
assert "flag" in tool.parameters["properties"]
def test_add_invalid_tool(self):
manager = ToolManager()
with pytest.raises(AttributeError):
manager.add_tool(1) # type: ignore
def test_add_lambda(self):
manager = ToolManager()
tool = manager.add_tool(lambda x: x, name="my_tool")
assert tool.name == "my_tool"
def test_add_lambda_with_no_name(self):
manager = ToolManager()
with pytest.raises(
ValueError, match="You must provide a name for lambda functions"
):
manager.add_tool(lambda x: x)
def test_warn_on_duplicate_tools(self, caplog):
"""Test warning on duplicate tools."""
def f(x: int) -> int:
return x
manager = ToolManager()
manager.add_tool(f)
with caplog.at_level(logging.WARNING):
manager.add_tool(f)
assert "Tool already exists: f" in caplog.text
def test_disable_warn_on_duplicate_tools(self, caplog):
"""Test disabling warning on duplicate tools."""
def f(x: int) -> int:
return x
manager = ToolManager()
manager.add_tool(f)
manager.warn_on_duplicate_tools = False
with caplog.at_level(logging.WARNING):
manager.add_tool(f)
assert "Tool already exists: f" not in caplog.text
class TestCallTools:
async def test_call_tool(self):
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
manager = ToolManager()
manager.add_tool(add)
result = await manager.call_tool("add", {"a": 1, "b": 2})
assert result == 3
async def test_call_async_tool(self):
async def double(n: int) -> int:
"""Double a number."""
return n * 2
manager = ToolManager()
manager.add_tool(double)
result = await manager.call_tool("double", {"n": 5})
assert result == 10
async def test_call_tool_with_default_args(self):
def add(a: int, b: int = 1) -> int:
"""Add two numbers."""
return a + b
manager = ToolManager()
manager.add_tool(add)
result = await manager.call_tool("add", {"a": 1})
assert result == 2
async def test_call_tool_with_missing_args(self):
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
manager = ToolManager()
manager.add_tool(add)
with pytest.raises(ToolError):
await manager.call_tool("add", {"a": 1})
async def test_call_unknown_tool(self):
manager = ToolManager()
with pytest.raises(ToolError):
await manager.call_tool("unknown", {"a": 1})
async def test_call_tool_with_list_int_input(self):
def sum_vals(vals: list[int]) -> int:
return sum(vals)
manager = ToolManager()
manager.add_tool(sum_vals)
# Try both with plain list and with JSON list
result = await manager.call_tool("sum_vals", {"vals": "[1, 2, 3]"})
assert result == 6
result = await manager.call_tool("sum_vals", {"vals": [1, 2, 3]})
assert result == 6
async def test_call_tool_with_list_str_or_str_input(self):
def concat_strs(vals: list[str] | str) -> str:
return vals if isinstance(vals, str) else "".join(vals)
manager = ToolManager()
manager.add_tool(concat_strs)
# Try both with plain python object and with JSON list
result = await manager.call_tool("concat_strs", {"vals": ["a", "b", "c"]})
assert result == "abc"
result = await manager.call_tool("concat_strs", {"vals": '["a", "b", "c"]'})
assert result == "abc"
result = await manager.call_tool("concat_strs", {"vals": "a"})
assert result == "a"
result = await manager.call_tool("concat_strs", {"vals": '"a"'})
assert result == '"a"'
async def test_call_tool_with_complex_model(self):
from fastmcp import Context
class MyShrimpTank(BaseModel):
class Shrimp(BaseModel):
name: str
shrimp: list[Shrimp]
x: None
def name_shrimp(tank: MyShrimpTank, ctx: Context) -> list[str]:
return [x.name for x in tank.shrimp]
manager = ToolManager()
manager.add_tool(name_shrimp)
result = await manager.call_tool(
"name_shrimp",
{"tank": {"x": None, "shrimp": [{"name": "rex"}, {"name": "gertrude"}]}},
)
assert result == ["rex", "gertrude"]
result = await manager.call_tool(
"name_shrimp",
{"tank": '{"x": null, "shrimp": [{"name": "rex"}, {"name": "gertrude"}]}'},
)
assert result == ["rex", "gertrude"]
class TestToolSchema:
async def test_context_arg_excluded_from_schema(self):
from fastmcp import Context
def something(a: int, ctx: Context) -> int:
return a
manager = ToolManager()
tool = manager.add_tool(something)
assert "ctx" not in json.dumps(tool.parameters)
assert "Context" not in json.dumps(tool.parameters)
assert "ctx" not in tool.fn_metadata.arg_model.model_fields
class TestContextHandling:
"""Test context handling in the tool manager."""
def test_context_parameter_detection(self):
"""Test that context parameters are properly detected in Tool.from_function()."""
from fastmcp import Context
def tool_with_context(x: int, ctx: Context) -> str:
return str(x)
manager = ToolManager()
tool = manager.add_tool(tool_with_context)
assert tool.context_kwarg == "ctx"
def tool_without_context(x: int) -> str:
return str(x)
tool = manager.add_tool(tool_without_context)
assert tool.context_kwarg is None
async def test_context_injection(self):
"""Test that context is properly injected during tool execution."""
from fastmcp import Context, FastMCP
def tool_with_context(x: int, ctx: Context) -> str:
assert isinstance(ctx, Context)
return str(x)
manager = ToolManager()
manager.add_tool(tool_with_context)
mcp = FastMCP()
ctx = mcp.get_context()
result = await manager.call_tool("tool_with_context", {"x": 42}, context=ctx)
assert result == "42"
async def test_context_injection_async(self):
"""Test that context is properly injected in async tools."""
from fastmcp import Context, FastMCP
async def async_tool(x: int, ctx: Context) -> str:
assert isinstance(ctx, Context)
return str(x)
manager = ToolManager()
manager.add_tool(async_tool)
mcp = FastMCP()
ctx = mcp.get_context()
result = await manager.call_tool("async_tool", {"x": 42}, context=ctx)
assert result == "42"
async def test_context_optional(self):
"""Test that context is optional when calling tools."""
from fastmcp import Context
def tool_with_context(x: int, ctx: Optional[Context] = None) -> str:
return str(x)
manager = ToolManager()
manager.add_tool(tool_with_context)
# Should not raise an error when context is not provided
result = await manager.call_tool("tool_with_context", {"x": 42})
assert result == "42"
async def test_context_error_handling(self):
"""Test error handling when context injection fails."""
from fastmcp import Context, FastMCP
def tool_with_context(x: int, ctx: Context) -> str:
raise ValueError("Test error")
manager = ToolManager()
manager.add_tool(tool_with_context)
mcp = FastMCP()
ctx = mcp.get_context()
with pytest.raises(ToolError, match="Error executing tool tool_with_context"):
await manager.call_tool("tool_with_context", {"x": 42}, context=ctx)

66
uv.lock generated
View file

@ -228,18 +228,14 @@ wheels = [
[[package]]
name = "fastmcp"
version = "0.3.6.dev8+g3b5ae20"
version = "0.4.2.dev4+g654397b.d20250401"
source = { editable = "." }
dependencies = [
{ name = "httpx" },
{ name = "mcp" },
{ name = "pydantic" },
{ name = "pydantic-settings" },
{ name = "python-dotenv" },
{ name = "typer" },
{ name = "rich" },
]
[package.optional-dependencies]
[package.dev-dependencies]
dev = [
{ name = "copychat" },
{ name = "ipython" },
@ -252,41 +248,25 @@ dev = [
{ name = "pytest-xdist" },
{ name = "ruff" },
]
tests = [
{ name = "pre-commit" },
{ name = "pyright" },
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-flakefinder" },
{ name = "pytest-xdist" },
{ name = "ruff" },
]
[package.metadata]
requires-dist = [
{ name = "copychat", marker = "extra == 'dev'", specifier = ">=0.5.2" },
{ name = "httpx", specifier = ">=0.26.0" },
{ name = "ipython", marker = "extra == 'dev'", specifier = ">=8.12.3" },
{ name = "mcp", specifier = ">=1.0.0,<2.0.0" },
{ name = "pdbpp", marker = "extra == 'dev'", specifier = ">=0.10.3" },
{ name = "pre-commit", marker = "extra == 'dev'" },
{ name = "pre-commit", marker = "extra == 'tests'" },
{ name = "pydantic", specifier = ">=2.5.3,<3.0.0" },
{ name = "pydantic-settings", specifier = ">=2.6.1" },
{ name = "pyright", marker = "extra == 'dev'", specifier = ">=1.1.389" },
{ name = "pyright", marker = "extra == 'tests'", specifier = ">=1.1.389" },
{ name = "pytest", marker = "extra == 'dev'", specifier = ">=8.3.3" },
{ name = "pytest", marker = "extra == 'tests'", specifier = ">=8.3.3" },
{ name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.23.5" },
{ name = "pytest-asyncio", marker = "extra == 'tests'", specifier = ">=0.23.5" },
{ name = "pytest-flakefinder", marker = "extra == 'dev'" },
{ name = "pytest-flakefinder", marker = "extra == 'tests'" },
{ name = "pytest-xdist", marker = "extra == 'dev'", specifier = ">=3.6.1" },
{ name = "pytest-xdist", marker = "extra == 'tests'", specifier = ">=3.6.1" },
{ name = "python-dotenv", specifier = ">=1.0.1" },
{ name = "ruff", marker = "extra == 'dev'" },
{ name = "ruff", marker = "extra == 'tests'" },
{ name = "typer", specifier = ">=0.9.0" },
{ name = "mcp", specifier = ">=1.6.0,<2.0.0" },
{ name = "rich", specifier = ">=13.9.4" },
]
[package.metadata.requires-dev]
dev = [
{ name = "copychat", specifier = ">=0.5.2" },
{ name = "ipython", specifier = ">=8.12.3" },
{ name = "pdbpp", specifier = ">=0.10.3" },
{ name = "pre-commit" },
{ name = "pyright", specifier = ">=1.1.389" },
{ name = "pytest", specifier = ">=8.3.3" },
{ name = "pytest-asyncio", specifier = ">=0.23.5" },
{ name = "pytest-flakefinder" },
{ name = "pytest-xdist", specifier = ">=3.6.1" },
{ name = "ruff" },
]
[[package]]
@ -455,19 +435,21 @@ wheels = [
[[package]]
name = "mcp"
version = "1.0.0"
version = "1.6.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
{ name = "httpx" },
{ name = "httpx-sse" },
{ name = "pydantic" },
{ name = "pydantic-settings" },
{ name = "sse-starlette" },
{ name = "starlette" },
{ name = "uvicorn" },
]
sdist = { url = "https://files.pythonhosted.org/packages/97/de/a9ec0a1b6439f90ea59f89004bb2e7ec6890dfaeef809751d9e6577dca7e/mcp-1.0.0.tar.gz", hash = "sha256:dba51ce0b5c6a80e25576f606760c49a91ee90210fed805b530ca165d3bbc9b7", size = 82891 }
sdist = { url = "https://files.pythonhosted.org/packages/95/d2/f587cb965a56e992634bebc8611c5b579af912b74e04eb9164bd49527d21/mcp-1.6.0.tar.gz", hash = "sha256:d9324876de2c5637369f43161cd71eebfd803df5a95e46225cab8d280e366723", size = 200031 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/56/89/900c0c8445ec001d3725e475fc553b0feb2e8a51be018f3bb7de51e683db/mcp-1.0.0-py3-none-any.whl", hash = "sha256:bbe70ffa3341cd4da78b5eb504958355c68381fb29971471cea1e642a2af5b8a", size = 36361 },
{ url = "https://files.pythonhosted.org/packages/10/30/20a7f33b0b884a9d14dd3aa94ff1ac9da1479fe2ad66dd9e2736075d2506/mcp-1.6.0-py3-none-any.whl", hash = "sha256:7bd24c6ea042dbec44c754f100984d186620d8b841ec30f1b19eda9b93a634d0", size = 76077 },
]
[[package]]