diff --git a/examples/fastapi_converter.py b/examples/fastapi_converter.py new file mode 100644 index 000000000..d1401c130 --- /dev/null +++ b/examples/fastapi_converter.py @@ -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) diff --git a/pyproject.toml b/pyproject.toml index 6fa835a4a..2a1c10806 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/src/fastmcp/__init__.py b/src/fastmcp/__init__.py index fdbfb9da4..955391205 100644 --- a/src/fastmcp/__init__.py +++ b/src/fastmcp/__init__.py @@ -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"] diff --git a/src/fastmcp/cli/__init__.py b/src/fastmcp/cli/__init__.py deleted file mode 100644 index 4de805855..000000000 --- a/src/fastmcp/cli/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -"""FastMCP CLI package.""" - -from .cli import app - - -if __name__ == "__main__": - app() diff --git a/src/fastmcp/cli/claude.py b/src/fastmcp/cli/claude.py deleted file mode 100644 index cb47eb23e..000000000 --- a/src/fastmcp/cli/claude.py +++ /dev/null @@ -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 diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py deleted file mode 100644 index de62efde3..000000000 --- a/src/fastmcp/cli/cli.py +++ /dev/null @@ -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) diff --git a/src/fastmcp/exceptions.py b/src/fastmcp/exceptions.py deleted file mode 100644 index fb5bda106..000000000 --- a/src/fastmcp/exceptions.py +++ /dev/null @@ -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.""" diff --git a/src/fastmcp/prompts/__init__.py b/src/fastmcp/prompts/__init__.py deleted file mode 100644 index 763726964..000000000 --- a/src/fastmcp/prompts/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -from .base import Prompt -from .manager import PromptManager - -__all__ = ["Prompt", "PromptManager"] diff --git a/src/fastmcp/prompts/base.py b/src/fastmcp/prompts/base.py deleted file mode 100644 index d44fc1823..000000000 --- a/src/fastmcp/prompts/base.py +++ /dev/null @@ -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 == "": - 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}") diff --git a/src/fastmcp/prompts/manager.py b/src/fastmcp/prompts/manager.py deleted file mode 100644 index f60e72cf9..000000000 --- a/src/fastmcp/prompts/manager.py +++ /dev/null @@ -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) diff --git a/src/fastmcp/prompts/prompt_manager.py b/src/fastmcp/prompts/prompt_manager.py deleted file mode 100644 index ea92a3b89..000000000 --- a/src/fastmcp/prompts/prompt_manager.py +++ /dev/null @@ -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()) diff --git a/src/fastmcp/resources/__init__.py b/src/fastmcp/resources/__init__.py deleted file mode 100644 index 92deb8735..000000000 --- a/src/fastmcp/resources/__init__.py +++ /dev/null @@ -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", -] diff --git a/src/fastmcp/resources/base.py b/src/fastmcp/resources/base.py deleted file mode 100644 index cf9c72b1b..000000000 --- a/src/fastmcp/resources/base.py +++ /dev/null @@ -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 diff --git a/src/fastmcp/resources/resource_manager.py b/src/fastmcp/resources/resource_manager.py deleted file mode 100644 index 5b8c3ad92..000000000 --- a/src/fastmcp/resources/resource_manager.py +++ /dev/null @@ -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()) diff --git a/src/fastmcp/resources/templates.py b/src/fastmcp/resources/templates.py deleted file mode 100644 index dc83730c8..000000000 --- a/src/fastmcp/resources/templates.py +++ /dev/null @@ -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 == "": - 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}") diff --git a/src/fastmcp/resources/types.py b/src/fastmcp/resources/types.py deleted file mode 100644 index 0b13816ec..000000000 --- a/src/fastmcp/resources/types.py +++ /dev/null @@ -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}") diff --git a/src/fastmcp/server.py b/src/fastmcp/server.py index 00bb21b0d..53286c782 100644 --- a/src/fastmcp/server.py +++ b/src/fastmcp/server.py @@ -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) diff --git a/src/fastmcp/settings.py b/src/fastmcp/settings.py new file mode 100644 index 000000000..fda707dd3 --- /dev/null +++ b/src/fastmcp/settings.py @@ -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", + ) diff --git a/src/fastmcp/tools/__init__.py b/src/fastmcp/tools/__init__.py deleted file mode 100644 index ae9c65619..000000000 --- a/src/fastmcp/tools/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -from .base import Tool -from .tool_manager import ToolManager - -__all__ = ["Tool", "ToolManager"] diff --git a/src/fastmcp/tools/base.py b/src/fastmcp/tools/base.py deleted file mode 100644 index 3b177d2bb..000000000 --- a/src/fastmcp/tools/base.py +++ /dev/null @@ -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 == "": - 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 diff --git a/src/fastmcp/tools/tool_manager.py b/src/fastmcp/tools/tool_manager.py deleted file mode 100644 index 2a4fca0ff..000000000 --- a/src/fastmcp/tools/tool_manager.py +++ /dev/null @@ -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) diff --git a/src/fastmcp/utilities/func_metadata.py b/src/fastmcp/utilities/func_metadata.py deleted file mode 100644 index dec156c9f..000000000 --- a/src/fastmcp/utilities/func_metadata.py +++ /dev/null @@ -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 diff --git a/tests/prompts/__init__.py b/tests/prompts/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/prompts/test_base.py b/tests/prompts/test_base.py deleted file mode 100644 index 4dca37d6d..000000000 --- a/tests/prompts/test_base.py +++ /dev/null @@ -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", - ), - ) - ) - ] diff --git a/tests/prompts/test_manager.py b/tests/prompts/test_manager.py deleted file mode 100644 index 823eaac1d..000000000 --- a/tests/prompts/test_manager.py +++ /dev/null @@ -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") diff --git a/tests/resources/__init__.py b/tests/resources/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/resources/test_file_resources.py b/tests/resources/test_file_resources.py deleted file mode 100644 index 83c8897a2..000000000 --- a/tests/resources/test_file_resources.py +++ /dev/null @@ -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 diff --git a/tests/resources/test_function_resources.py b/tests/resources/test_function_resources.py deleted file mode 100644 index 3a2d5e5e2..000000000 --- a/tests/resources/test_function_resources.py +++ /dev/null @@ -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) diff --git a/tests/resources/test_resource_manager.py b/tests/resources/test_resource_manager.py deleted file mode 100644 index 87061d911..000000000 --- a/tests/resources/test_resource_manager.py +++ /dev/null @@ -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] diff --git a/tests/resources/test_resource_template.py b/tests/resources/test_resource_template.py deleted file mode 100644 index 9b459d7b1..000000000 --- a/tests/resources/test_resource_template.py +++ /dev/null @@ -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" diff --git a/tests/resources/test_resources.py b/tests/resources/test_resources.py deleted file mode 100644 index 9eb3d3721..000000000 --- a/tests/resources/test_resources.py +++ /dev/null @@ -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 diff --git a/tests/servers/__init__.py b/tests/servers/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/tests/servers/test_file_server.py b/tests/servers/test_file_server.py deleted file mode 100644 index 1eb750eee..000000000 --- a/tests/servers/test_file_server.py +++ /dev/null @@ -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" diff --git a/tests/test_cli.py b/tests/test_cli.py deleted file mode 100644 index fefd6ca20..000000000 --- a/tests/test_cli.py +++ /dev/null @@ -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() diff --git a/tests/test_func_metadata.py b/tests/test_func_metadata.py deleted file mode 100644 index eebcb0401..000000000 --- a/tests/test_func_metadata.py +++ /dev/null @@ -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": "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", - } diff --git a/tests/test_server.py b/tests/test_server.py deleted file mode 100644 index 6303cbaea..000000000 --- a/tests/test_server.py +++ /dev/null @@ -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") diff --git a/tests/test_tool_manager.py b/tests/test_tool_manager.py deleted file mode 100644 index 4356a9a22..000000000 --- a/tests/test_tool_manager.py +++ /dev/null @@ -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) diff --git a/uv.lock b/uv.lock index 3b07278c5..50c733d79 100644 --- a/uv.lock +++ b/uv.lock @@ -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]]