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

Move OpenAPI to providers/openapi submodule (#2672)

Browse source 
* Move OpenAPI to providers/openapi/ submodule

Migrates the OpenAPI implementation to the provider pattern:
- Creates src/fastmcp/server/providers/openapi/ submodule with provider.py,
  components.py, and routing.py
- Converts server/openapi/ to deprecated re-export stubs
- FastMCPOpenAPI remains as deprecated wrapper that uses OpenAPIProvider
- Updates experimental/server/openapi to import from new canonical location

* Update tests and FastMCP methods to use OpenAPIProvider

- Update FastMCP.from_openapi() and from_fastapi() to use OpenAPIProvider
  directly, returning FastMCP instead of deprecated FastMCPOpenAPI
- Move tests from tests/server/openapi/ to tests/server/providers/openapi/
- Update all test imports to use new canonical location
- Add tests/deprecated/openapi/ for backward compatibility testing
- Update tests/client/test_openapi.py to use new imports

* Fix deprecation warning tests and address CodeRabbit feedback

- Use importlib.reload() to ensure deprecation warnings fire in tests
- Change logger.error to logger.exception for better stack traces
This commit is contained in:
Jeremiah Lowin 2025-12-22 11:13:33 -05:00 • committed by GitHub
commit 96c7a74b27
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 1692 additions and 1417 deletions
Download patch file Download diff file Expand all files Collapse all files

33
src/fastmcp/experimental/server/openapi/__init__.py
View file

@ -1,28 +1,31 @@
"""Deprecated: Import from fastmcp.server.openapi instead."""
"""Deprecated: Import from fastmcp.server.providers.openapi instead."""
import warnings
from fastmcp.server.openapi import (
ComponentFn,
DEFAULT_ROUTE_MAPPINGS,
FastMCPOpenAPI,
MCPType,
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
RouteMap,
RouteMapFn,
_determine_route_type,
)
# Deprecated in 2.14 when OpenAPI support was promoted out of experimental
warnings.warn(
"Importing from fastmcp.experimental.server.openapi is deprecated. "
"Import from fastmcp.server.openapi instead.",
"Import from fastmcp.server.providers.openapi instead.",
DeprecationWarning,
stacklevel=2,
)
# Import from canonical location
from fastmcp.server.openapi.server import FastMCPOpenAPI as FastMCPOpenAPI # noqa: E402
from fastmcp.server.providers.openapi import ( # noqa: E402
ComponentFn as ComponentFn,
MCPType as MCPType,
OpenAPIResource as OpenAPIResource,
OpenAPIResourceTemplate as OpenAPIResourceTemplate,
OpenAPITool as OpenAPITool,
RouteMap as RouteMap,
RouteMapFn as RouteMapFn,
)
from fastmcp.server.providers.openapi.routing import ( # noqa: E402
DEFAULT_ROUTE_MAPPINGS as DEFAULT_ROUTE_MAPPINGS,
_determine_route_type as _determine_route_type,
)
__all__ = [
"DEFAULT_ROUTE_MAPPINGS",
"ComponentFn",

58
src/fastmcp/server/openapi/__init__.py
View file

@ -1,35 +1,55 @@
"""OpenAPI server implementation for FastMCP - refactored for better maintainability."""
"""OpenAPI server implementation for FastMCP.
# Import from server
from .server import FastMCPOpenAPI
.. deprecated::
This module is deprecated. Import from fastmcp.server.providers.openapi instead.
# Import from routing
from .routing import (
MCPType,
RouteMap,
RouteMapFn,
ComponentFn,
DEFAULT_ROUTE_MAPPINGS,
_determine_route_type,
The recommended approach is to use OpenAPIProvider with FastMCP:
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
import httpx
client = httpx.AsyncClient(base_url="https://api.example.com")
provider = OpenAPIProvider(openapi_spec=spec, client=client)
mcp = FastMCP("My API Server")
mcp.add_provider(provider)
FastMCPOpenAPI is still available but deprecated.
"""
import warnings
warnings.warn(
"fastmcp.server.openapi is deprecated. "
"Import from fastmcp.server.providers.openapi instead.",
DeprecationWarning,
stacklevel=2,
)
# Import from components
from .components import (
OpenAPITool,
OpenAPIResource,
OpenAPIResourceTemplate,
# Re-export from new canonical location
from fastmcp.server.providers.openapi import ( # noqa: E402
ComponentFn as ComponentFn,
MCPType as MCPType,
OpenAPIProvider as OpenAPIProvider,
OpenAPIResource as OpenAPIResource,
OpenAPIResourceTemplate as OpenAPIResourceTemplate,
OpenAPITool as OpenAPITool,
RouteMap as RouteMap,
RouteMapFn as RouteMapFn,
)
# Export public symbols - maintaining backward compatibility
# Keep FastMCPOpenAPI for backwards compat (it has its own deprecation warning)
from fastmcp.server.openapi.server import FastMCPOpenAPI as FastMCPOpenAPI # noqa: E402
__all__ = [
"DEFAULT_ROUTE_MAPPINGS",
"ComponentFn",
"FastMCPOpenAPI",
"MCPType",
"OpenAPIProvider",
"OpenAPIResource",
"OpenAPIResourceTemplate",
"OpenAPITool",
"RouteMap",
"RouteMapFn",
"_determine_route_type",
]

361
src/fastmcp/server/openapi/components.py
View file

@ -1,353 +1,24 @@
"""OpenAPI component implementations: Tool, Resource, and ResourceTemplate classes."""
"""OpenAPI component implementations - backwards compatibility stub.
import json
import re
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
This module is deprecated. Import from fastmcp.server.providers.openapi instead.
"""
import httpx
from mcp.types import ToolAnnotations
from pydantic.networks import AnyUrl
from __future__ import annotations
from fastmcp.resources import Resource, ResourceContent, ResourceTemplate
from fastmcp.server.dependencies import get_http_headers
from fastmcp.tools.tool import Tool, ToolResult
from fastmcp.utilities.logging import get_logger
import warnings
# Import from our new utilities
from fastmcp.utilities.openapi import HTTPRoute
from fastmcp.utilities.openapi.director import RequestDirector
if TYPE_CHECKING:
from fastmcp.server import Context
logger = get_logger(__name__)
class OpenAPITool(Tool):
"""Tool implementation for OpenAPI endpoints."""
def __init__(
self,
client: httpx.AsyncClient,
route: HTTPRoute,
director: RequestDirector,
name: str,
description: str,
parameters: dict[str, Any],
output_schema: dict[str, Any] | None = None,
tags: set[str] | None = None,
timeout: float | None = None,
annotations: ToolAnnotations | None = None,
serializer: Callable[[Any], str] | None = None,
):
super().__init__(
name=name,
description=description,
parameters=parameters,
output_schema=output_schema,
tags=tags or set(),
annotations=annotations,
serializer=serializer,
)
self._client = client
self._route = route
self._director = director
self._timeout = timeout
def __repr__(self) -> str:
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPITool(name={self.name!r}, method={self._route.method}, path={self._route.path})"
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""Execute the HTTP request using RequestDirector for simplified parameter handling."""
try:
# Get base URL from client
base_url = (
str(self._client.base_url)
if hasattr(self._client, "base_url") and self._client.base_url
else "http://localhost"
)
# Get Headers from client
cli_headers = (
self._client.headers
if hasattr(self._client, "headers") and self._client.headers
else {}
)
# Build the request using RequestDirector
request = self._director.build(self._route, arguments, base_url)
# First add server headers (lowest precedence)
if cli_headers:
# Merge with existing headers, _client headers as base
if request.headers:
# Start with request headers, then add client headers
for key, value in cli_headers.items():
if key not in request.headers:
request.headers[key] = value
else:
# Create new headers from cli_headers
for key, value in cli_headers.items():
request.headers[key] = value
# Then add MCP client transport headers (highest precedence)
mcp_headers = get_http_headers()
if mcp_headers:
# Merge with existing headers, MCP headers take precedence over all
if request.headers:
request.headers.update(mcp_headers)
else:
# Create new headers from mcp_headers
for key, value in mcp_headers.items():
request.headers[key] = value
# print logger
logger.debug(f"run - sending request; headers: {request.headers}")
# Execute the request
# Note: httpx.AsyncClient.send() doesn't accept timeout parameter
# The timeout should be configured on the client itself
response = await self._client.send(request)
# Raise for 4xx/5xx responses
response.raise_for_status()
# Try to parse as JSON first
try:
result = response.json()
# Handle structured content based on output schema, if any
structured_output = None
if self.output_schema is not None:
if self.output_schema.get("x-fastmcp-wrap-result"):
# Schema says wrap - always wrap in result key
structured_output = {"result": result}
else:
structured_output = result
# If no output schema, use fallback logic for backward compatibility
elif not isinstance(result, dict):
structured_output = {"result": result}
else:
structured_output = result
return ToolResult(structured_content=structured_output)
except json.JSONDecodeError:
return ToolResult(content=response.text)
except httpx.HTTPStatusError as e:
# Handle HTTP errors (4xx, 5xx)
error_message = (
f"HTTP error {e.response.status_code}: {e.response.reason_phrase}"
)
try:
error_data = e.response.json()
error_message += f" - {error_data}"
except (json.JSONDecodeError, ValueError):
if e.response.text:
error_message += f" - {e.response.text}"
raise ValueError(error_message) from e
except httpx.RequestError as e:
# Handle request errors (connection, timeout, etc.)
raise ValueError(f"Request error: {e!s}") from e
class OpenAPIResource(Resource):
"""Resource implementation for OpenAPI endpoints."""
def __init__(
self,
client: httpx.AsyncClient,
route: HTTPRoute,
director: RequestDirector,
uri: str,
name: str,
description: str,
mime_type: str = "application/json",
tags: set[str] | None = None,
timeout: float | None = None,
):
if tags is None:
tags = set()
super().__init__(
uri=AnyUrl(uri), # Convert string to AnyUrl
name=name,
description=description,
mime_type=mime_type,
tags=tags,
)
self._client = client
self._route = route
self._director = director
self._timeout = timeout
def __repr__(self) -> str:
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPIResource(name={self.name!r}, uri={self.uri!r}, path={self._route.path})"
async def read(self) -> ResourceContent:
"""Fetch the resource data by making an HTTP request."""
try:
# Extract path parameters from the URI if present
path = self._route.path
resource_uri = str(self.uri)
# If this is a templated resource, extract path parameters from the URI
if "{" in path and "}" in path:
# Extract the resource ID from the URI (the last part after the last slash)
parts = resource_uri.split("/")
if len(parts) > 1:
# Find all path parameters in the route path
path_params = {}
# Find the path parameter names from the route path
param_matches = re.findall(r"\{([^}]+)\}", path)
if param_matches:
# Reverse sorting from creation order (traversal is backwards)
param_matches.sort(reverse=True)
# Number of sent parameters is number of parts -1 (assuming first part is resource identifier)
expected_param_count = len(parts) - 1
# Map parameters from the end of the URI to the parameters in the path
# Last parameter in URI (parts[-1]) maps to last parameter in path, and so on
for i, param_name in enumerate(param_matches):
# Ensure we don't use resource identifier as parameter
if i < expected_param_count:
# Get values from the end of parts
param_value = parts[-1 - i]
path_params[param_name] = param_value
# Replace path parameters with their values
for param_name, param_value in path_params.items():
path = path.replace(f"{{{param_name}}}", str(param_value))
# Filter any query parameters - get query parameters and filter out None/empty values
query_params = {}
for param in self._route.parameters:
if param.location == "query" and hasattr(self, f"_{param.name}"):
value = getattr(self, f"_{param.name}")
if value is not None and value != "":
query_params[param.name] = value
# Prepare headers with correct precedence: server < client transport
headers = {}
# Start with server headers (lowest precedence)
cli_headers = (
self._client.headers
if hasattr(self._client, "headers") and self._client.headers
else {}
)
headers.update(cli_headers)
# Add MCP client transport headers (highest precedence)
mcp_headers = get_http_headers()
headers.update(mcp_headers)
response = await self._client.request(
method=self._route.method,
url=path,
params=query_params,
headers=headers,
timeout=self._timeout,
)
# Raise for 4xx/5xx responses
response.raise_for_status()
# Determine content type and return appropriate format
content_type = response.headers.get("content-type", "").lower()
if "application/json" in content_type:
result = response.json()
return ResourceContent(
content=json.dumps(result), mime_type="application/json"
)
elif any(ct in content_type for ct in ["text/", "application/xml"]):
return ResourceContent(content=response.text, mime_type=self.mime_type)
else:
return ResourceContent(
content=response.content, mime_type=self.mime_type
)
except httpx.HTTPStatusError as e:
# Handle HTTP errors (4xx, 5xx)
error_message = (
f"HTTP error {e.response.status_code}: {e.response.reason_phrase}"
)
try:
error_data = e.response.json()
error_message += f" - {error_data}"
except (json.JSONDecodeError, ValueError):
if e.response.text:
error_message += f" - {e.response.text}"
raise ValueError(error_message) from e
except httpx.RequestError as e:
# Handle request errors (connection, timeout, etc.)
raise ValueError(f"Request error: {e!s}") from e
class OpenAPIResourceTemplate(ResourceTemplate):
"""Resource template implementation for OpenAPI endpoints."""
def __init__(
self,
client: httpx.AsyncClient,
route: HTTPRoute,
director: RequestDirector,
uri_template: str,
name: str,
description: str,
parameters: dict[str, Any],
tags: set[str] | None = None,
timeout: float | None = None,
):
if tags is None:
tags = set()
super().__init__(
uri_template=uri_template,
name=name,
description=description,
parameters=parameters,
tags=tags,
)
self._client = client
self._route = route
self._director = director
self._timeout = timeout
def __repr__(self) -> str:
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPIResourceTemplate(name={self.name!r}, uri_template={self.uri_template!r}, path={self._route.path})"
async def create_resource(
self,
uri: str,
params: dict[str, Any],
context: "Context | None" = None,
) -> Resource:
"""Create a resource with the given parameters."""
# Generate a URI for this resource instance
uri_parts = []
for key, value in params.items():
uri_parts.append(f"{key}={value}")
# Create and return a resource
return OpenAPIResource(
client=self._client,
route=self._route,
director=self._director,
uri=uri,
name=f"{self.name}-{'-'.join(uri_parts)}",
description=self.description or f"Resource for {self._route.path}",
mime_type="application/json",
tags=set(self._route.tags or []),
timeout=self._timeout,
)
warnings.warn(
"fastmcp.server.openapi.components is deprecated. "
"Import from fastmcp.server.providers.openapi instead.",
DeprecationWarning,
stacklevel=2,
)
from fastmcp.server.providers.openapi import ( # noqa: E402
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
)
# Export public symbols
__all__ = [

152
src/fastmcp/server/openapi/routing.py
View file

@ -1,125 +1,14 @@
"""Route mapping logic for OpenAPI operations."""
"""Route mapping logic for OpenAPI operations.
import enum
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from re import Pattern
from typing import TYPE_CHECKING, Literal
.. deprecated::
This module is deprecated. Import from fastmcp.server.providers.openapi instead.
"""
if TYPE_CHECKING:
from .components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
)
# Import from our new utilities
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.openapi import HttpMethod, HTTPRoute
# ruff: noqa: E402
logger = get_logger(__name__)
import warnings
# Type definitions for the mapping functions
RouteMapFn = Callable[[HTTPRoute, "MCPType"], "MCPType | None"]
ComponentFn = Callable[
[
HTTPRoute,
"OpenAPITool | OpenAPIResource | OpenAPIResourceTemplate",
],
None,
]
class MCPType(enum.Enum):
"""Type of FastMCP component to create from a route.
Enum values:
TOOL: Convert the route to a callable Tool
RESOURCE: Convert the route to a Resource (typically GET endpoints)
RESOURCE_TEMPLATE: Convert the route to a ResourceTemplate (typically GET with path params)
EXCLUDE: Exclude the route from being converted to any MCP component
"""
TOOL = "TOOL"
RESOURCE = "RESOURCE"
RESOURCE_TEMPLATE = "RESOURCE_TEMPLATE"
# PROMPT = "PROMPT"
EXCLUDE = "EXCLUDE"
@dataclass(kw_only=True)
class RouteMap:
"""Mapping configuration for HTTP routes to FastMCP component types."""
methods: list[HttpMethod] | Literal["*"] = field(default="*")
pattern: Pattern[str] | str = field(default=r".*")
tags: set[str] = field(
default_factory=set,
metadata={"description": "A set of tags to match. All tags must match."},
)
mcp_type: MCPType = field(
metadata={"description": "The type of FastMCP component to create."},
)
mcp_tags: set[str] = field(
default_factory=set,
metadata={
"description": "A set of tags to apply to the generated FastMCP component."
},
)
# Default route mapping: all routes become tools.
# Users can provide custom route_maps to override this behavior.
DEFAULT_ROUTE_MAPPINGS = [
RouteMap(mcp_type=MCPType.TOOL),
]
def _determine_route_type(
route: HTTPRoute,
mappings: list[RouteMap],
) -> RouteMap:
"""
Determines the FastMCP component type based on the route and mappings.
Args:
route: HTTPRoute object
mappings: List of RouteMap objects in priority order
Returns:
The RouteMap that matches the route, or a catchall "Tool" RouteMap if no match is found.
"""
# Check mappings in priority order (first match wins)
for route_map in mappings:
# Check if the HTTP method matches
if route_map.methods == "*" or route.method in route_map.methods:
# Handle both string patterns and compiled Pattern objects
if isinstance(route_map.pattern, Pattern):
pattern_matches = route_map.pattern.search(route.path)
else:
pattern_matches = re.search(route_map.pattern, route.path)
if pattern_matches:
# Check if tags match (if specified)
# If route_map.tags is empty, tags are not matched
# If route_map.tags is non-empty, all tags must be present in route.tags (AND condition)
if route_map.tags:
route_tags_set = set(route.tags or [])
if not route_map.tags.issubset(route_tags_set):
# Tags don't match, continue to next mapping
continue
logger.debug(
f"Route {route.method} {route.path} mapped to {route_map.mcp_type.name}"
)
return route_map
# Default fallback
return RouteMap(mcp_type=MCPType.TOOL)
# Export public symbols
# Backwards compatibility - export everything that was previously public
__all__ = [
"DEFAULT_ROUTE_MAPPINGS",
"ComponentFn",
@ -128,3 +17,30 @@ __all__ = [
"RouteMapFn",
"_determine_route_type",
]
warnings.warn(
"fastmcp.server.openapi.routing is deprecated. "
"Import from fastmcp.server.providers.openapi instead.",
DeprecationWarning,
stacklevel=2,
)
# Re-export from new canonical location
from fastmcp.server.providers.openapi.routing import (
DEFAULT_ROUTE_MAPPINGS as DEFAULT_ROUTE_MAPPINGS,
)
from fastmcp.server.providers.openapi.routing import (
ComponentFn as ComponentFn,
)
from fastmcp.server.providers.openapi.routing import (
MCPType as MCPType,
)
from fastmcp.server.providers.openapi.routing import (
RouteMap as RouteMap,
)
from fastmcp.server.providers.openapi.routing import (
RouteMapFn as RouteMapFn,
)
from fastmcp.server.providers.openapi.routing import (
_determine_route_type as _determine_route_type,
)

466
src/fastmcp/server/openapi/server.py
View file

@ -1,100 +1,64 @@
"""FastMCP server implementation for OpenAPI integration."""
"""FastMCPOpenAPI - backwards compatibility wrapper.
import re
from collections import Counter
from typing import Any, Literal
This class is deprecated. Use FastMCP with OpenAPIProvider instead:
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
import httpx
client = httpx.AsyncClient(base_url="https://api.example.com")
provider = OpenAPIProvider(openapi_spec=spec, client=client)
mcp = FastMCP("My API Server")
mcp.add_provider(provider)
"""
from __future__ import annotations
import warnings
from typing import Any
import httpx
from jsonschema_path import SchemaPath
from fastmcp.server.server import FastMCP
from fastmcp.utilities.logging import get_logger
# Import from our new utilities and components
from fastmcp.utilities.openapi import (
HTTPRoute,
extract_output_schema_from_responses,
format_simple_description,
parse_openapi_to_http_routes,
)
from fastmcp.utilities.openapi.director import RequestDirector
from .components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
)
from .routing import (
DEFAULT_ROUTE_MAPPINGS,
from fastmcp.server.providers.openapi import (
ComponentFn,
MCPType,
OpenAPIProvider,
RouteMap,
RouteMapFn,
_determine_route_type,
)
logger = get_logger(__name__)
def _slugify(text: str) -> str:
"""
Convert text to a URL-friendly slug format that only contains lowercase
letters, uppercase letters, numbers, and underscores.
"""
if not text:
return ""
# Replace spaces and common separators with underscores
slug = re.sub(r"[\s\-\.]+", "_", text)
# Remove non-alphanumeric characters except underscores
slug = re.sub(r"[^a-zA-Z0-9_]", "", slug)
# Remove multiple consecutive underscores
slug = re.sub(r"_+", "_", slug)
# Remove leading/trailing underscores
slug = slug.strip("_")
return slug
from fastmcp.server.server import FastMCP
class FastMCPOpenAPI(FastMCP):
"""
FastMCP server implementation that creates components from an OpenAPI schema.
"""FastMCP server implementation that creates components from an OpenAPI schema.
This class parses an OpenAPI specification and creates appropriate FastMCP components
(Tools, Resources, ResourceTemplates) based on route mappings.
.. deprecated::
Use FastMCP with OpenAPIProvider instead. This class will be
removed in a future version.
Example:
Example (deprecated):
```python
from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap, MCPType
from fastmcp.server.openapi import FastMCPOpenAPI
import httpx
# Define custom route mappings
custom_mappings = [
# Map all user-related endpoints to ResourceTemplate
RouteMap(
methods=["GET", "POST", "PATCH"],
pattern=r".*/users/.*",
mcp_type=MCPType.RESOURCE_TEMPLATE
),
# Map all analytics endpoints to Tool
RouteMap(
methods=["GET"],
pattern=r".*/analytics/.*",
mcp_type=MCPType.TOOL
),
]
# Create server with custom mappings and route mapper
server = FastMCPOpenAPI(
openapi_spec=spec,
client=httpx.AsyncClient(),
name="API Server",
route_maps=custom_mappings,
)
```
New approach:
```python
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
import httpx
client = httpx.AsyncClient(base_url="https://api.example.com")
provider = OpenAPIProvider(openapi_spec=spec, client=client)
mcp = FastMCP("API Server")
mcp.add_provider(provider)
```
"""
def __init__(
@ -110,340 +74,56 @@ class FastMCPOpenAPI(FastMCP):
timeout: float | None = None,
**settings: Any,
):
"""
Initialize a FastMCP server from an OpenAPI schema.
"""Initialize a FastMCP server from an OpenAPI schema.
.. deprecated::
Use FastMCP with OpenAPIProvider instead.
Args:
openapi_spec: OpenAPI schema as a dictionary or file path
openapi_spec: OpenAPI schema as a dictionary
client: httpx AsyncClient for making HTTP requests
name: Optional name for the server
route_maps: Optional list of RouteMap objects defining route mappings
route_map_fn: Optional callable for advanced route type mapping.
Receives (route, mcp_type) and returns MCPType or None.
Called on every route, including excluded ones.
mcp_component_fn: Optional callable for component customization.
Receives (route, component) and can modify the component in-place.
Called on every created component.
mcp_names: Optional dictionary mapping operationId to desired component names.
If an operationId is not in the dictionary, falls back to using the
operationId up to the first double underscore. If no operationId exists,
falls back to slugified summary or path-based naming.
All names are truncated to 56 characters maximum.
tags: Optional set of tags to add to all components. Components always receive any tags
from the route.
route_map_fn: Optional callable for advanced route type mapping
mcp_component_fn: Optional callable for component customization
mcp_names: Optional dictionary mapping operationId to component names
tags: Optional set of tags to add to all components
timeout: Optional timeout (in seconds) for all requests
**settings: Additional settings for FastMCP
"""
warnings.warn(
"FastMCPOpenAPI is deprecated. Use FastMCP with OpenAPIProvider instead:\n"
" provider = OpenAPIProvider(openapi_spec=spec, client=client)\n"
" mcp = FastMCP('name')\n"
" mcp.add_provider(provider)",
DeprecationWarning,
stacklevel=2,
)
super().__init__(name=name or "OpenAPI FastMCP", **settings)
# Store references for backwards compatibility
self._client = client
self._timeout = timeout
self._mcp_component_fn = mcp_component_fn
# Keep track of names to detect collisions
self._used_names = {
"tool": Counter(),
"resource": Counter(),
"resource_template": Counter(),
"prompt": Counter(),
}
# Create openapi-core Spec and RequestDirector for stateless request building
try:
self._spec = SchemaPath.from_dict(openapi_spec) # type: ignore[arg-type]
self._director = RequestDirector(self._spec)
except Exception as e:
logger.error(f"Failed to initialize RequestDirector: {e}")
raise ValueError(f"Invalid OpenAPI specification: {e}") from e
http_routes = parse_openapi_to_http_routes(openapi_spec)
# Process routes
route_maps = (route_maps or []) + DEFAULT_ROUTE_MAPPINGS
for route in http_routes:
# Determine route type based on mappings or default rules
route_map = _determine_route_type(route, route_maps)
route_type = route_map.mcp_type
# Call route_map_fn if provided
if route_map_fn is not None:
try:
result = route_map_fn(route, route_type)
if result is not None:
route_type = result
logger.debug(
f"Route {route.method} {route.path} mapping customized by route_map_fn: "
f"type={route_type.name}"
)
except Exception as e:
logger.warning(
f"Error in route_map_fn for {route.method} {route.path}: {e}. "
f"Using default values."
)
# Generate a default name from the route
component_name = self._generate_default_name(route, mcp_names)
route_tags = set(route.tags) | route_map.mcp_tags | (tags or set())
# Create components using simplified approach with RequestDirector
if route_type == MCPType.TOOL:
self._create_openapi_tool(route, component_name, tags=route_tags)
elif route_type == MCPType.RESOURCE:
self._create_openapi_resource(route, component_name, tags=route_tags)
elif route_type == MCPType.RESOURCE_TEMPLATE:
self._create_openapi_template(route, component_name, tags=route_tags)
elif route_type == MCPType.EXCLUDE:
logger.debug(f"Excluding route: {route.method} {route.path}")
logger.debug(f"Created FastMCP OpenAPI server with {len(http_routes)} routes")
def _generate_default_name(
self, route: HTTPRoute, mcp_names_map: dict[str, str] | None = None
) -> str:
"""Generate a default name from the route using the configured strategy."""
name = ""
mcp_names_map = mcp_names_map or {}
# First check if there's a custom mapping for this operationId
if route.operation_id:
if route.operation_id in mcp_names_map:
name = mcp_names_map[route.operation_id]
else:
# If there's a double underscore in the operationId, use the first part
name = route.operation_id.split("__")[0]
else:
name = route.summary or f"{route.method}_{route.path}"
name = _slugify(name)
# Truncate to 56 characters maximum
if len(name) > 56:
name = name[:56]
return name
def _get_unique_name(
self,
name: str,
component_type: Literal["tool", "resource", "resource_template", "prompt"],
) -> str:
"""
Ensure the name is unique within its component type by appending numbers if needed.
Args:
name: The proposed name
component_type: The type of component ("tools", "resources", or "templates")
Returns:
str: A unique name for the component
"""
# Check if the name is already used
self._used_names[component_type][name] += 1
if self._used_names[component_type][name] == 1:
return name
else:
# Create the new name
new_name = f"{name}_{self._used_names[component_type][name]}"
logger.debug(
f"Name collision detected: '{name}' already exists as a {component_type}. "
f"Using '{new_name}' instead."
)
return new_name
def _create_openapi_tool(
self,
route: HTTPRoute,
name: str,
tags: set[str],
):
"""Creates and registers an OpenAPITool with enhanced description."""
# Use pre-calculated schema from route
combined_schema = route.flat_param_schema
# Extract output schema from OpenAPI responses
output_schema = extract_output_schema_from_responses(
route.responses,
route.response_schemas,
route.openapi_version,
# Create provider with the client
provider = OpenAPIProvider(
openapi_spec=openapi_spec,
client=client,
route_maps=route_maps,
route_map_fn=route_map_fn,
mcp_component_fn=mcp_component_fn,
mcp_names=mcp_names,
tags=tags,
timeout=timeout,
)
# Get a unique tool name
tool_name = self._get_unique_name(name, "tool")
self.add_provider(provider)
base_description = (
route.description
or route.summary
or f"Executes {route.method} {route.path}"
)
# Use simplified description formatter for tools
enhanced_description = format_simple_description(
base_description=base_description,
parameters=route.parameters,
request_body=route.request_body,
)
tool = OpenAPITool(
client=self._client,
route=route,
director=self._director,
name=tool_name,
description=enhanced_description,
parameters=combined_schema,
output_schema=output_schema,
tags=set(route.tags or []) | tags,
timeout=self._timeout,
)
# Call component_fn if provided
if self._mcp_component_fn is not None:
try:
self._mcp_component_fn(route, tool)
logger.debug(f"Tool {tool_name} customized by component_fn")
except Exception as e:
logger.warning(
f"Error in component_fn for tool {tool_name}: {e}. "
f"Using component as-is."
)
# Use the potentially modified tool name as the registration key
final_tool_name = tool.name
# Register the tool by directly assigning to the tools dictionary
self._tool_manager._tools[final_tool_name] = tool
def _create_openapi_resource(
self,
route: HTTPRoute,
name: str,
tags: set[str],
):
"""Creates and registers an OpenAPIResource with enhanced description."""
# Get a unique resource name
resource_name = self._get_unique_name(name, "resource")
resource_uri = f"resource://{resource_name}"
base_description = (
route.description or route.summary or f"Represents {route.path}"
)
# Use simplified description for resources
enhanced_description = format_simple_description(
base_description=base_description,
parameters=route.parameters,
request_body=route.request_body,
)
resource = OpenAPIResource(
client=self._client,
route=route,
director=self._director,
uri=resource_uri,
name=resource_name,
description=enhanced_description,
tags=set(route.tags or []) | tags,
timeout=self._timeout,
)
# Call component_fn if provided
if self._mcp_component_fn is not None:
try:
self._mcp_component_fn(route, resource)
logger.debug(f"Resource {resource_uri} customized by component_fn")
except Exception as e:
logger.warning(
f"Error in component_fn for resource {resource_uri}: {e}. "
f"Using component as-is."
)
# Use the potentially modified resource URI as the registration key
final_resource_uri = str(resource.uri)
# Register the resource by directly assigning to the resources dictionary
self._resource_manager._resources[final_resource_uri] = resource
def _create_openapi_template(
self,
route: HTTPRoute,
name: str,
tags: set[str],
):
"""Creates and registers an OpenAPIResourceTemplate with enhanced description."""
# Get a unique template name
template_name = self._get_unique_name(name, "resource_template")
path_params = [p.name for p in route.parameters if p.location == "path"]
path_params.sort() # Sort for consistent URIs
uri_template_str = f"resource://{template_name}"
if path_params:
uri_template_str += "/" + "/".join(f"{{{p}}}" for p in path_params)
base_description = (
route.description or route.summary or f"Template for {route.path}"
)
# Use simplified description for resource templates
enhanced_description = format_simple_description(
base_description=base_description,
parameters=route.parameters,
request_body=route.request_body,
)
template_params_schema = {
"type": "object",
"properties": {
p.name: {
**(p.schema_.copy() if isinstance(p.schema_, dict) else {}),
**(
{"description": p.description}
if p.description
and not (
isinstance(p.schema_, dict) and "description" in p.schema_
)
else {}
),
}
for p in route.parameters
if p.location == "path"
},
"required": [
p.name for p in route.parameters if p.location == "path" and p.required
],
}
template = OpenAPIResourceTemplate(
client=self._client,
route=route,
director=self._director,
uri_template=uri_template_str,
name=template_name,
description=enhanced_description,
parameters=template_params_schema,
tags=set(route.tags or []) | tags,
timeout=self._timeout,
)
# Call component_fn if provided
if self._mcp_component_fn is not None:
try:
self._mcp_component_fn(route, template)
logger.debug(f"Template {uri_template_str} customized by component_fn")
except Exception as e:
logger.warning(
f"Error in component_fn for template {uri_template_str}: {e}. "
f"Using component as-is."
)
# Use the potentially modified template URI as the registration key
final_template_uri = template.uri_template
# Register the template by directly assigning to the templates dictionary
self._resource_manager._templates[final_template_uri] = template
# Expose internal attributes for backwards compatibility
self._spec = provider._spec
self._director = provider._director
# Export public symbols

8
src/fastmcp/server/providers/__init__.py
View file

@ -32,10 +32,12 @@ from fastmcp.server.providers.fastmcp_provider import FastMCPProvider
from fastmcp.server.providers.transforming import TransformingProvider
if TYPE_CHECKING:
from fastmcp.server.providers.openapi import OpenAPIProvider as OpenAPIProvider
from fastmcp.server.providers.proxy import ProxyProvider as ProxyProvider
__all__ = [
"FastMCPProvider",
"OpenAPIProvider",
"Provider",
"ProxyProvider",
"TransformingProvider",
@ -43,9 +45,13 @@ __all__ = [
def __getattr__(name: str):
"""Lazy import for ProxyProvider to avoid circular imports."""
"""Lazy import for providers to avoid circular imports."""
if name == "ProxyProvider":
from fastmcp.server.providers.proxy import ProxyProvider
return ProxyProvider
if name == "OpenAPIProvider":
from fastmcp.server.providers.openapi import OpenAPIProvider
return OpenAPIProvider
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

0
src/fastmcp/server/openapi/README.md → src/fastmcp/server/providers/openapi/README.md
View file

41
src/fastmcp/server/providers/openapi/__init__.py Normal file
View file

@ -0,0 +1,41 @@
"""OpenAPI provider for FastMCP.
This module provides OpenAPI integration for FastMCP through the Provider pattern.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
import httpx
client = httpx.AsyncClient(base_url="https://api.example.com")
provider = OpenAPIProvider(openapi_spec=spec, client=client)
mcp = FastMCP("API Server")
mcp.add_provider(provider)
```
"""
from fastmcp.server.providers.openapi.components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
)
from fastmcp.server.providers.openapi.provider import OpenAPIProvider
from fastmcp.server.providers.openapi.routing import (
ComponentFn,
MCPType,
RouteMap,
RouteMapFn,
)
__all__ = [
"ComponentFn",
"MCPType",
"OpenAPIProvider",
"OpenAPIResource",
"OpenAPIResourceTemplate",
"OpenAPITool",
"RouteMap",
"RouteMapFn",
]

309
src/fastmcp/server/providers/openapi/components.py Normal file
View file

@ -0,0 +1,309 @@
"""OpenAPI component classes: Tool, Resource, and ResourceTemplate."""
from __future__ import annotations
import json
import re
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
import httpx
from mcp.types import ToolAnnotations
from pydantic.networks import AnyUrl
from fastmcp.resources import Resource, ResourceContent, ResourceTemplate
from fastmcp.server.dependencies import get_http_headers
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.tools.tool import Tool, ToolResult
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.openapi import HTTPRoute
from fastmcp.utilities.openapi.director import RequestDirector
if TYPE_CHECKING:
from fastmcp.server import Context
__all__ = [
"OpenAPIResource",
"OpenAPIResourceTemplate",
"OpenAPITool",
]
logger = get_logger(__name__)
def _slugify(text: str) -> str:
"""Convert text to a URL-friendly slug format.
Only contains lowercase letters, uppercase letters, numbers, and underscores.
"""
if not text:
return ""
# Replace spaces and common separators with underscores
slug = re.sub(r"[\s\-\.]+", "_", text)
# Remove non-alphanumeric characters except underscores
slug = re.sub(r"[^a-zA-Z0-9_]", "", slug)
# Remove multiple consecutive underscores
slug = re.sub(r"_+", "_", slug)
# Remove leading/trailing underscores
slug = slug.strip("_")
return slug
class OpenAPITool(Tool):
"""Tool implementation for OpenAPI endpoints."""
task_config: TaskConfig = TaskConfig(mode="forbidden")
def __init__(
self,
client: httpx.AsyncClient,
route: HTTPRoute,
director: RequestDirector,
name: str,
description: str,
parameters: dict[str, Any],
output_schema: dict[str, Any] | None = None,
tags: set[str] | None = None,
timeout: float | None = None,
annotations: ToolAnnotations | None = None,
serializer: Callable[[Any], str] | None = None,
):
super().__init__(
name=name,
description=description,
parameters=parameters,
output_schema=output_schema,
tags=tags or set(),
annotations=annotations,
serializer=serializer,
)
self._client = client
self._route = route
self._director = director
self._timeout = timeout
def __repr__(self) -> str:
return f"OpenAPITool(name={self.name!r}, method={self._route.method}, path={self._route.path})"
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""Execute the HTTP request using RequestDirector."""
try:
base_url = (
str(self._client.base_url)
if self._client.base_url
else "http://localhost"
)
# Build the request using RequestDirector
request = self._director.build(self._route, arguments, base_url)
# Add client headers (lowest precedence)
if self._client.headers:
for key, value in self._client.headers.items():
if key not in request.headers:
request.headers[key] = value
# Add MCP transport headers (highest precedence)
mcp_headers = get_http_headers()
if mcp_headers:
request.headers.update(mcp_headers)
logger.debug(f"run - sending request; headers: {request.headers}")
response = await self._client.send(request)
response.raise_for_status()
# Try to parse as JSON first
try:
result = response.json()
# Handle structured content based on output schema
if self.output_schema is not None:
if self.output_schema.get("x-fastmcp-wrap-result"):
structured_output = {"result": result}
else:
structured_output = result
elif not isinstance(result, dict):
structured_output = {"result": result}
else:
structured_output = result
return ToolResult(structured_content=structured_output)
except json.JSONDecodeError:
return ToolResult(content=response.text)
except httpx.HTTPStatusError as e:
error_message = (
f"HTTP error {e.response.status_code}: {e.response.reason_phrase}"
)
try:
error_data = e.response.json()
error_message += f" - {error_data}"
except (json.JSONDecodeError, ValueError):
if e.response.text:
error_message += f" - {e.response.text}"
raise ValueError(error_message) from e
except httpx.RequestError as e:
raise ValueError(f"Request error: {e!s}") from e
class OpenAPIResource(Resource):
"""Resource implementation for OpenAPI endpoints."""
task_config: TaskConfig = TaskConfig(mode="forbidden")
def __init__(
self,
client: httpx.AsyncClient,
route: HTTPRoute,
director: RequestDirector,
uri: str,
name: str,
description: str,
mime_type: str = "application/json",
tags: set[str] | None = None,
timeout: float | None = None,
):
super().__init__(
uri=AnyUrl(uri),
name=name,
description=description,
mime_type=mime_type,
tags=tags or set(),
)
self._client = client
self._route = route
self._director = director
self._timeout = timeout
def __repr__(self) -> str:
return f"OpenAPIResource(name={self.name!r}, uri={self.uri!r}, path={self._route.path})"
async def read(self) -> ResourceContent:
"""Fetch the resource data by making an HTTP request."""
try:
path = self._route.path
resource_uri = str(self.uri)
# If this is a templated resource, extract path parameters from the URI
if "{" in path and "}" in path:
parts = resource_uri.split("/")
if len(parts) > 1:
path_params = {}
param_matches = re.findall(r"\{([^}]+)\}", path)
if param_matches:
param_matches.sort(reverse=True)
expected_param_count = len(parts) - 1
for i, param_name in enumerate(param_matches):
if i < expected_param_count:
param_value = parts[-1 - i]
path_params[param_name] = param_value
for param_name, param_value in path_params.items():
path = path.replace(f"{{{param_name}}}", str(param_value))
# Build headers with correct precedence
headers: dict[str, str] = {}
if self._client.headers:
headers.update(self._client.headers)
mcp_headers = get_http_headers()
if mcp_headers:
headers.update(mcp_headers)
response = await self._client.request(
method=self._route.method,
url=path,
headers=headers,
timeout=self._timeout,
)
response.raise_for_status()
content_type = response.headers.get("content-type", "").lower()
if "application/json" in content_type:
result = response.json()
return ResourceContent(
content=json.dumps(result), mime_type="application/json"
)
elif any(ct in content_type for ct in ["text/", "application/xml"]):
return ResourceContent(content=response.text, mime_type=self.mime_type)
else:
return ResourceContent(
content=response.content, mime_type=self.mime_type
)
except httpx.HTTPStatusError as e:
error_message = (
f"HTTP error {e.response.status_code}: {e.response.reason_phrase}"
)
try:
error_data = e.response.json()
error_message += f" - {error_data}"
except (json.JSONDecodeError, ValueError):
if e.response.text:
error_message += f" - {e.response.text}"
raise ValueError(error_message) from e
except httpx.RequestError as e:
raise ValueError(f"Request error: {e!s}") from e
class OpenAPIResourceTemplate(ResourceTemplate):
"""Resource template implementation for OpenAPI endpoints."""
task_config: TaskConfig = TaskConfig(mode="forbidden")
def __init__(
self,
client: httpx.AsyncClient,
route: HTTPRoute,
director: RequestDirector,
uri_template: str,
name: str,
description: str,
parameters: dict[str, Any],
tags: set[str] | None = None,
timeout: float | None = None,
):
super().__init__(
uri_template=uri_template,
name=name,
description=description,
parameters=parameters,
tags=tags or set(),
)
self._client = client
self._route = route
self._director = director
self._timeout = timeout
def __repr__(self) -> str:
return f"OpenAPIResourceTemplate(name={self.name!r}, uri_template={self.uri_template!r}, path={self._route.path})"
async def create_resource(
self,
uri: str,
params: dict[str, Any],
context: Context | None = None,
) -> Resource:
"""Create a resource with the given parameters."""
uri_parts = [f"{key}={value}" for key, value in params.items()]
return OpenAPIResource(
client=self._client,
route=self._route,
director=self._director,
uri=uri,
name=f"{self.name}-{'-'.join(uri_parts)}",
description=self.description or f"Resource for {self._route.path}",
mime_type="application/json",
tags=set(self._route.tags or []),
timeout=self._timeout,
)

383
src/fastmcp/server/providers/openapi/provider.py Normal file
View file

@ -0,0 +1,383 @@
"""OpenAPIProvider for creating MCP components from OpenAPI specifications."""
from __future__ import annotations
from collections import Counter
from collections.abc import Sequence
from typing import Any, Literal
import httpx
from jsonschema_path import SchemaPath
from fastmcp.prompts import Prompt
from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.server.providers.base import Provider, TaskComponents
from fastmcp.server.providers.openapi.components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
_slugify,
)
from fastmcp.server.providers.openapi.routing import (
DEFAULT_ROUTE_MAPPINGS,
ComponentFn,
MCPType,
RouteMap,
RouteMapFn,
_determine_route_type,
)
from fastmcp.tools.tool import Tool
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.openapi import (
HTTPRoute,
extract_output_schema_from_responses,
format_simple_description,
parse_openapi_to_http_routes,
)
from fastmcp.utilities.openapi.director import RequestDirector
__all__ = [
"OpenAPIProvider",
]
logger = get_logger(__name__)
class OpenAPIProvider(Provider):
"""Provider that creates MCP components from an OpenAPI specification.
Components are created eagerly during initialization by parsing the OpenAPI
spec. Each component makes HTTP calls to the described API endpoints.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
import httpx
client = httpx.AsyncClient(base_url="https://api.example.com")
provider = OpenAPIProvider(openapi_spec=spec, client=client)
mcp = FastMCP("API Server")
mcp.add_provider(provider)
```
"""
def __init__(
self,
openapi_spec: dict[str, Any],
client: httpx.AsyncClient,
*,
route_maps: list[RouteMap] | None = None,
route_map_fn: RouteMapFn | None = None,
mcp_component_fn: ComponentFn | None = None,
mcp_names: dict[str, str] | None = None,
tags: set[str] | None = None,
timeout: float | None = None,
):
"""Initialize provider by parsing OpenAPI spec and creating components.
Args:
openapi_spec: OpenAPI schema as a dictionary
client: httpx AsyncClient for making HTTP requests
route_maps: Optional list of RouteMap objects defining route mappings
route_map_fn: Optional callable for advanced route type mapping
mcp_component_fn: Optional callable for component customization
mcp_names: Optional dictionary mapping operationId to component names
tags: Optional set of tags to add to all components
timeout: Optional timeout (in seconds) for all requests
"""
super().__init__()
self._client = client
self._timeout = timeout
self._mcp_component_fn = mcp_component_fn
# Keep track of names to detect collisions
self._used_names: dict[str, Counter[str]] = {
"tool": Counter(),
"resource": Counter(),
"resource_template": Counter(),
"prompt": Counter(),
}
# Pre-created component storage
self._tools: dict[str, OpenAPITool] = {}
self._resources: dict[str, OpenAPIResource] = {}
self._templates: dict[str, OpenAPIResourceTemplate] = {}
# Create openapi-core Spec and RequestDirector
try:
self._spec = SchemaPath.from_dict(openapi_spec) # type: ignore[arg-type]
self._director = RequestDirector(self._spec)
except Exception as e:
logger.exception("Failed to initialize RequestDirector")
raise ValueError(f"Invalid OpenAPI specification: {e}") from e
http_routes = parse_openapi_to_http_routes(openapi_spec)
# Process routes
route_maps = (route_maps or []) + DEFAULT_ROUTE_MAPPINGS
for route in http_routes:
route_map = _determine_route_type(route, route_maps)
route_type = route_map.mcp_type
if route_map_fn is not None:
try:
result = route_map_fn(route, route_type)
if result is not None:
route_type = result
logger.debug(
f"Route {route.method} {route.path} mapping customized: "
f"type={route_type.name}"
)
except Exception as e:
logger.warning(
f"Error in route_map_fn for {route.method} {route.path}: {e}. "
f"Using default values."
)
component_name = self._generate_default_name(route, mcp_names)
route_tags = set(route.tags) | route_map.mcp_tags | (tags or set())
if route_type == MCPType.TOOL:
self._create_openapi_tool(route, component_name, tags=route_tags)
elif route_type == MCPType.RESOURCE:
self._create_openapi_resource(route, component_name, tags=route_tags)
elif route_type == MCPType.RESOURCE_TEMPLATE:
self._create_openapi_template(route, component_name, tags=route_tags)
elif route_type == MCPType.EXCLUDE:
logger.debug(f"Excluding route: {route.method} {route.path}")
logger.debug(f"Created OpenAPIProvider with {len(http_routes)} routes")
def _generate_default_name(
self, route: HTTPRoute, mcp_names_map: dict[str, str] | None = None
) -> str:
"""Generate a default name from the route."""
mcp_names_map = mcp_names_map or {}
if route.operation_id:
if route.operation_id in mcp_names_map:
name = mcp_names_map[route.operation_id]
else:
name = route.operation_id.split("__")[0]
else:
name = route.summary or f"{route.method}_{route.path}"
name = _slugify(name)
if len(name) > 56:
name = name[:56]
return name
def _get_unique_name(
self,
name: str,
component_type: Literal["tool", "resource", "resource_template", "prompt"],
) -> str:
"""Ensure the name is unique by appending numbers if needed."""
self._used_names[component_type][name] += 1
if self._used_names[component_type][name] == 1:
return name
new_name = f"{name}_{self._used_names[component_type][name]}"
logger.debug(
f"Name collision: '{name}' exists as {component_type}. Using '{new_name}'."
)
return new_name
def _create_openapi_tool(
self,
route: HTTPRoute,
name: str,
tags: set[str],
) -> None:
"""Create and register an OpenAPITool."""
combined_schema = route.flat_param_schema
output_schema = extract_output_schema_from_responses(
route.responses,
route.response_schemas,
route.openapi_version,
)
tool_name = self._get_unique_name(name, "tool")
base_description = (
route.description
or route.summary
or f"Executes {route.method} {route.path}"
)
enhanced_description = format_simple_description(
base_description=base_description,
parameters=route.parameters,
request_body=route.request_body,
)
tool = OpenAPITool(
client=self._client,
route=route,
director=self._director,
name=tool_name,
description=enhanced_description,
parameters=combined_schema,
output_schema=output_schema,
tags=set(route.tags or []) | tags,
timeout=self._timeout,
)
if self._mcp_component_fn is not None:
try:
self._mcp_component_fn(route, tool)
logger.debug(f"Tool {tool_name} customized by component_fn")
except Exception as e:
logger.warning(f"Error in component_fn for tool {tool_name}: {e}")
self._tools[tool.name] = tool
def _create_openapi_resource(
self,
route: HTTPRoute,
name: str,
tags: set[str],
) -> None:
"""Create and register an OpenAPIResource."""
resource_name = self._get_unique_name(name, "resource")
resource_uri = f"resource://{resource_name}"
base_description = (
route.description or route.summary or f"Represents {route.path}"
)
enhanced_description = format_simple_description(
base_description=base_description,
parameters=route.parameters,
request_body=route.request_body,
)
resource = OpenAPIResource(
client=self._client,
route=route,
director=self._director,
uri=resource_uri,
name=resource_name,
description=enhanced_description,
tags=set(route.tags or []) | tags,
timeout=self._timeout,
)
if self._mcp_component_fn is not None:
try:
self._mcp_component_fn(route, resource)
logger.debug(f"Resource {resource_uri} customized by component_fn")
except Exception as e:
logger.warning(
f"Error in component_fn for resource {resource_uri}: {e}"
)
self._resources[str(resource.uri)] = resource
def _create_openapi_template(
self,
route: HTTPRoute,
name: str,
tags: set[str],
) -> None:
"""Create and register an OpenAPIResourceTemplate."""
template_name = self._get_unique_name(name, "resource_template")
path_params = sorted(p.name for p in route.parameters if p.location == "path")
uri_template_str = f"resource://{template_name}"
if path_params:
uri_template_str += "/" + "/".join(f"{{{p}}}" for p in path_params)
base_description = (
route.description or route.summary or f"Template for {route.path}"
)
enhanced_description = format_simple_description(
base_description=base_description,
parameters=route.parameters,
request_body=route.request_body,
)
template_params_schema = {
"type": "object",
"properties": {
p.name: {
**(p.schema_.copy() if isinstance(p.schema_, dict) else {}),
**(
{"description": p.description}
if p.description
and not (
isinstance(p.schema_, dict) and "description" in p.schema_
)
else {}
),
}
for p in route.parameters
if p.location == "path"
},
"required": [
p.name for p in route.parameters if p.location == "path" and p.required
],
}
template = OpenAPIResourceTemplate(
client=self._client,
route=route,
director=self._director,
uri_template=uri_template_str,
name=template_name,
description=enhanced_description,
parameters=template_params_schema,
tags=set(route.tags or []) | tags,
timeout=self._timeout,
)
if self._mcp_component_fn is not None:
try:
self._mcp_component_fn(route, template)
logger.debug(f"Template {uri_template_str} customized by component_fn")
except Exception as e:
logger.warning(
f"Error in component_fn for template {uri_template_str}: {e}"
)
self._templates[template.uri_template] = template
# -------------------------------------------------------------------------
# Provider interface
# -------------------------------------------------------------------------
async def list_tools(self) -> Sequence[Tool]:
"""Return all tools created from the OpenAPI spec."""
return list(self._tools.values())
async def get_tool(self, name: str) -> Tool | None:
"""Get a tool by name."""
return self._tools.get(name)
async def list_resources(self) -> Sequence[Resource]:
"""Return all resources created from the OpenAPI spec."""
return list(self._resources.values())
async def get_resource(self, uri: str) -> Resource | None:
"""Get a resource by URI."""
return self._resources.get(uri)
async def list_resource_templates(self) -> Sequence[ResourceTemplate]:
"""Return all resource templates created from the OpenAPI spec."""
return list(self._templates.values())
async def get_resource_template(self, uri: str) -> ResourceTemplate | None:
"""Get a resource template that matches the given URI."""
return next(
(t for t in self._templates.values() if t.matches(uri) is not None),
None,
)
async def list_prompts(self) -> Sequence[Prompt]:
"""Return empty list - OpenAPI doesn't create prompts."""
return []
async def get_tasks(self) -> TaskComponents:
"""Return empty TaskComponents - OpenAPI components don't support tasks."""
return TaskComponents()

109
src/fastmcp/server/providers/openapi/routing.py Normal file
View file

@ -0,0 +1,109 @@
"""Route mapping logic for OpenAPI operations."""
from __future__ import annotations
import enum
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from re import Pattern
from typing import TYPE_CHECKING, Literal
if TYPE_CHECKING:
from fastmcp.server.providers.openapi.components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
)
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.openapi import HttpMethod, HTTPRoute
__all__ = [
"ComponentFn",
"MCPType",
"RouteMap",
"RouteMapFn",
]
logger = get_logger(__name__)
# Type definitions for the mapping functions
RouteMapFn = Callable[[HTTPRoute, "MCPType"], "MCPType | None"]
ComponentFn = Callable[
[
HTTPRoute,
"OpenAPITool | OpenAPIResource | OpenAPIResourceTemplate",
],
None,
]
class MCPType(enum.Enum):
"""Type of FastMCP component to create from a route.
Enum values:
TOOL: Convert the route to a callable Tool
RESOURCE: Convert the route to a Resource (typically GET endpoints)
RESOURCE_TEMPLATE: Convert the route to a ResourceTemplate (typically GET with path params)
EXCLUDE: Exclude the route from being converted to any MCP component
"""
TOOL = "TOOL"
RESOURCE = "RESOURCE"
RESOURCE_TEMPLATE = "RESOURCE_TEMPLATE"
EXCLUDE = "EXCLUDE"
@dataclass(kw_only=True)
class RouteMap:
"""Mapping configuration for HTTP routes to FastMCP component types."""
methods: list[HttpMethod] | Literal["*"] = field(default="*")
pattern: Pattern[str] | str = field(default=r".*")
tags: set[str] = field(
default_factory=set,
metadata={"description": "A set of tags to match. All tags must match."},
)
mcp_type: MCPType = field(
metadata={"description": "The type of FastMCP component to create."},
)
mcp_tags: set[str] = field(
default_factory=set,
metadata={
"description": "A set of tags to apply to the generated FastMCP component."
},
)
# Default route mapping: all routes become tools.
DEFAULT_ROUTE_MAPPINGS = [
RouteMap(mcp_type=MCPType.TOOL),
]
def _determine_route_type(
route: HTTPRoute,
mappings: list[RouteMap],
) -> RouteMap:
"""Determine the FastMCP component type based on the route and mappings."""
for route_map in mappings:
if route_map.methods == "*" or route.method in route_map.methods:
if isinstance(route_map.pattern, Pattern):
pattern_matches = route_map.pattern.search(route.path)
else:
pattern_matches = re.search(route_map.pattern, route.path)
if pattern_matches:
if route_map.tags:
route_tags_set = set(route.tags or [])
if not route_map.tags.issubset(route_tags_set):
continue
logger.debug(
f"Route {route.method} {route.path} mapped to {route_map.mcp_type.name}"
)
return route_map
return RouteMap(mcp_type=MCPType.TOOL)

67
src/fastmcp/server/server.py
View file

@ -52,6 +52,7 @@ from starlette.middleware import Middleware as ASGIMiddleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import BaseRoute, Route
from typing_extensions import Self
import fastmcp
import fastmcp.server
@ -96,9 +97,9 @@ if TYPE_CHECKING:
from fastmcp.client.client import FastMCP1Server
from fastmcp.client.sampling import SamplingHandler
from fastmcp.client.transports import ClientTransport, ClientTransportT
from fastmcp.server.openapi import ComponentFn as OpenAPIComponentFn
from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap
from fastmcp.server.openapi import RouteMapFn as OpenAPIRouteMapFn
from fastmcp.server.providers.openapi import ComponentFn as OpenAPIComponentFn
from fastmcp.server.providers.openapi import RouteMap
from fastmcp.server.providers.openapi import RouteMapFn as OpenAPIRouteMapFn
from fastmcp.server.providers.proxy import FastMCPProxy
from fastmcp.tools.tool import ToolResultSerializerType
@ -2748,19 +2749,36 @@ class FastMCP(Generic[LifespanResultT]):
cls,
openapi_spec: dict[str, Any],
client: httpx.AsyncClient,
name: str = "OpenAPI Server",
route_maps: list[RouteMap] | None = None,
route_map_fn: OpenAPIRouteMapFn | None = None,
mcp_component_fn: OpenAPIComponentFn | None = None,
mcp_names: dict[str, str] | None = None,
tags: set[str] | None = None,
timeout: float | None = None,
**settings: Any,
) -> FastMCPOpenAPI:
) -> Self:
"""
Create a FastMCP server from an OpenAPI specification.
"""
from .openapi import FastMCPOpenAPI
return FastMCPOpenAPI(
Args:
openapi_spec: OpenAPI schema as a dictionary
client: httpx AsyncClient for making HTTP requests
name: Name for the MCP server
route_maps: Optional list of RouteMap objects defining route mappings
route_map_fn: Optional callable for advanced route type mapping
mcp_component_fn: Optional callable for component customization
mcp_names: Optional dictionary mapping operationId to component names
tags: Optional set of tags to add to all components
timeout: Optional timeout (in seconds) for all requests
**settings: Additional settings passed to FastMCP
Returns:
A FastMCP server with an OpenAPIProvider attached.
"""
from .providers.openapi import OpenAPIProvider
provider = OpenAPIProvider(
openapi_spec=openapi_spec,
client=client,
route_maps=route_maps,
@ -2768,8 +2786,11 @@ class FastMCP(Generic[LifespanResultT]):
mcp_component_fn=mcp_component_fn,
mcp_names=mcp_names,
tags=tags,
**settings,
timeout=timeout,
)
server = cls(name=name, **settings)
server.add_provider(provider)
return server
@classmethod
def from_fastapi(
@ -2782,12 +2803,28 @@ class FastMCP(Generic[LifespanResultT]):
mcp_names: dict[str, str] | None = None,
httpx_client_kwargs: dict[str, Any] | None = None,
tags: set[str] | None = None,
timeout: float | None = None,
**settings: Any,
) -> FastMCPOpenAPI:
) -> Self:
"""
Create a FastMCP server from a FastAPI application.
Args:
app: FastAPI application instance
name: Name for the MCP server (defaults to app.title)
route_maps: Optional list of RouteMap objects defining route mappings
route_map_fn: Optional callable for advanced route type mapping
mcp_component_fn: Optional callable for component customization
mcp_names: Optional dictionary mapping operationId to component names
httpx_client_kwargs: Optional kwargs passed to httpx.AsyncClient
tags: Optional set of tags to add to all components
timeout: Optional timeout (in seconds) for all requests
**settings: Additional settings passed to FastMCP
Returns:
A FastMCP server with an OpenAPIProvider attached.
"""
from .openapi import FastMCPOpenAPI
from .providers.openapi import OpenAPIProvider
if httpx_client_kwargs is None:
httpx_client_kwargs = {}
@ -2798,19 +2835,21 @@ class FastMCP(Generic[LifespanResultT]):
**httpx_client_kwargs,
)
name = name or app.title
server_name = name or app.title
return FastMCPOpenAPI(
provider = OpenAPIProvider(
openapi_spec=app.openapi(),
client=client,
name=name,
route_maps=route_maps,
route_map_fn=route_map_fn,
mcp_component_fn=mcp_component_fn,
mcp_names=mcp_names,
tags=tags,
**settings,
timeout=timeout,
)
server = cls(name=server_name, **settings)
server.add_provider(provider)
return server
@classmethod
def as_proxy(

2
tests/client/test_openapi.py
View file

@ -6,7 +6,7 @@ from mcp.types import TextResourceContents
from fastmcp import Client, FastMCP
from fastmcp.client.transports import SSETransport, StreamableHttpTransport
from fastmcp.server.openapi import MCPType, RouteMap
from fastmcp.server.providers.openapi import MCPType, RouteMap
from fastmcp.utilities.tests import run_server_async

170
tests/deprecated/openapi/test_openapi.py Normal file
View file

@ -0,0 +1,170 @@
"""Tests for deprecated OpenAPI imports.
These tests verify that the old import paths still work and emit
deprecation warnings, ensuring backwards compatibility.
"""
import warnings
import httpx
class TestDeprecatedServerOpenAPIImports:
"""Test deprecated imports from fastmcp.server.openapi."""
def test_import_fastmcp_openapi_emits_warning(self):
"""Importing from fastmcp.server.openapi should emit deprecation warning."""
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
# Force reimport
import importlib
import fastmcp.server.openapi
importlib.reload(fastmcp.server.openapi)
deprecation_warnings = [
x for x in w if issubclass(x.category, DeprecationWarning)
]
assert len(deprecation_warnings) >= 1
assert "providers.openapi" in str(deprecation_warnings[0].message)
def test_import_routing_emits_warning(self):
"""Importing from fastmcp.server.openapi.routing should emit deprecation warning."""
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
import importlib
import fastmcp.server.openapi.routing
importlib.reload(fastmcp.server.openapi.routing)
deprecation_warnings = [
x for x in w if issubclass(x.category, DeprecationWarning)
]
assert len(deprecation_warnings) >= 1
assert "providers.openapi" in str(deprecation_warnings[0].message)
def test_fastmcp_openapi_class_emits_warning(self):
"""Using FastMCPOpenAPI should emit deprecation warning."""
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
from fastmcp.server.openapi.server import FastMCPOpenAPI
spec = {
"openapi": "3.0.0",
"info": {"title": "Test", "version": "1.0.0"},
"paths": {},
}
client = httpx.AsyncClient(base_url="https://example.com")
FastMCPOpenAPI(openapi_spec=spec, client=client)
deprecation_warnings = [
x for x in w if issubclass(x.category, DeprecationWarning)
]
assert len(deprecation_warnings) >= 1
assert "FastMCPOpenAPI" in str(deprecation_warnings[-1].message)
def test_deprecated_imports_still_work(self):
"""All expected symbols should be importable from deprecated locations."""
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
from fastmcp.server.openapi import (
FastMCPOpenAPI,
MCPType,
OpenAPIProvider,
RouteMap,
)
# Verify they're the right types
assert FastMCPOpenAPI is not None
assert OpenAPIProvider is not None
assert MCPType.TOOL.value == "TOOL"
assert RouteMap is not None
def test_deprecated_routing_imports_still_work(self):
"""Routing symbols should be importable from deprecated location."""
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
from fastmcp.server.openapi.routing import (
DEFAULT_ROUTE_MAPPINGS,
MCPType,
_determine_route_type,
)
assert DEFAULT_ROUTE_MAPPINGS is not None
assert len(DEFAULT_ROUTE_MAPPINGS) > 0
assert MCPType.TOOL.value == "TOOL"
assert _determine_route_type is not None
class TestDeprecatedExperimentalOpenAPIImports:
"""Test deprecated imports from fastmcp.experimental.server.openapi."""
def test_experimental_import_emits_warning(self):
"""Importing from experimental should emit deprecation warning."""
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
import importlib
import fastmcp.experimental.server.openapi
importlib.reload(fastmcp.experimental.server.openapi)
deprecation_warnings = [
x for x in w if issubclass(x.category, DeprecationWarning)
]
assert len(deprecation_warnings) >= 1
assert "providers.openapi" in str(deprecation_warnings[0].message)
def test_experimental_imports_still_work(self):
"""All expected symbols should be importable from experimental."""
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
from fastmcp.experimental.server.openapi import (
DEFAULT_ROUTE_MAPPINGS,
FastMCPOpenAPI,
MCPType,
)
assert FastMCPOpenAPI is not None
assert DEFAULT_ROUTE_MAPPINGS is not None
assert MCPType.TOOL.value == "TOOL"
class TestDeprecatedComponentsImports:
"""Test deprecated imports from fastmcp.server.openapi.components."""
def test_components_import_emits_warning(self):
"""Importing from components should emit deprecation warning."""
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
import importlib
import fastmcp.server.openapi.components
importlib.reload(fastmcp.server.openapi.components)
deprecation_warnings = [
x for x in w if issubclass(x.category, DeprecationWarning)
]
assert len(deprecation_warnings) >= 1
assert "providers.openapi" in str(deprecation_warnings[0].message)
def test_components_imports_still_work(self):
"""Component classes should be importable from deprecated location."""
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
from fastmcp.server.openapi.components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
)
assert OpenAPITool is not None
assert OpenAPIResource is not None
assert OpenAPIResourceTemplate is not None

9
tests/deprecated/test_openapi_deprecations.py
View file

@ -1,5 +1,6 @@
"""Tests for OpenAPI-related deprecations in 2.14."""
import importlib
import warnings
import pytest
@ -37,19 +38,23 @@ class TestExperimentalOpenAPIImportDeprecation:
def test_experimental_server_openapi_import_warns(self):
"""Importing from fastmcp.experimental.server.openapi should warn."""
import fastmcp.experimental.server.openapi
with pytest.warns(
DeprecationWarning,
match=r"Importing from fastmcp\.experimental\.server\.openapi is deprecated",
):
from fastmcp.experimental.server.openapi import FastMCPOpenAPI # noqa: F401
importlib.reload(fastmcp.experimental.server.openapi)
def test_experimental_utilities_openapi_import_warns(self):
"""Importing from fastmcp.experimental.utilities.openapi should warn."""
import fastmcp.experimental.utilities.openapi
with pytest.warns(
DeprecationWarning,
match=r"Importing from fastmcp\.experimental\.utilities\.openapi is deprecated",
):
from fastmcp.experimental.utilities.openapi import HTTPRoute # noqa: F401
importlib.reload(fastmcp.experimental.utilities.openapi)
def test_experimental_imports_resolve_to_same_classes(self):
"""Experimental imports should resolve to the same classes as main imports."""

325
tests/server/openapi/test_end_to_end_compatibility.py
View file

@ -1,325 +0,0 @@
"""End-to-end compatibility tests between legacy and new OpenAPI implementations."""
import httpx
import pytest
from fastmcp.client import Client
from fastmcp.server.openapi import FastMCPOpenAPI
from fastmcp.server.openapi import FastMCPOpenAPI as LegacyFastMCPOpenAPI
class TestEndToEndCompatibility:
"""Test that legacy and new implementations create identical tools."""
@pytest.fixture
def simple_spec(self):
"""Simple OpenAPI spec for testing."""
return {
"openapi": "3.0.0",
"info": {"title": "Test API", "version": "1.0.0"},
"paths": {
"/users/{id}": {
"get": {
"operationId": "get_user",
"summary": "Get user by ID",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "integer"},
},
{
"name": "include_details",
"in": "query",
"required": False,
"schema": {"type": "boolean"},
},
],
"responses": {"200": {"description": "User found"}},
}
}
},
}
@pytest.fixture
def collision_spec(self):
"""OpenAPI spec with parameter collisions."""
return {
"openapi": "3.0.0",
"info": {"title": "Collision API", "version": "1.0.0"},
"paths": {
"/users/{id}": {
"put": {
"operationId": "update_user",
"summary": "Update user",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "integer"},
}
],
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
},
"required": ["name"],
}
}
},
},
"responses": {"200": {"description": "User updated"}},
}
}
},
}
async def test_tool_schema_compatibility(self, simple_spec):
"""Test that tools have identical input schemas."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
# Create both servers
legacy_server = LegacyFastMCPOpenAPI(
openapi_spec=simple_spec,
client=client,
name="Legacy Server",
)
new_server = FastMCPOpenAPI(
openapi_spec=simple_spec,
client=client,
name="New Server",
)
# Get tools from both servers
async with Client(legacy_server) as legacy_client:
legacy_tools = await legacy_client.list_tools()
async with Client(new_server) as new_client:
new_tools = await new_client.list_tools()
# Should have same number of tools
assert len(legacy_tools) == len(new_tools)
assert len(legacy_tools) == 1
# Get the single tool from each
legacy_tool = legacy_tools[0]
new_tool = new_tools[0]
# Names should be identical
assert legacy_tool.name == new_tool.name
assert legacy_tool.name == "get_user"
# Descriptions may differ (new server has simplified descriptions)
# Just check that both have descriptions
assert legacy_tool.description
assert new_tool.description
# Input schemas should be identical
legacy_schema = legacy_tool.inputSchema
new_schema = new_tool.inputSchema
# Required fields should match
assert set(legacy_schema.get("required", [])) == set(
new_schema.get("required", [])
)
# Properties should match
legacy_props = legacy_schema.get("properties", {})
new_props = new_schema.get("properties", {})
assert set(legacy_props.keys()) == set(new_props.keys())
# Check each property
for prop_name in legacy_props:
legacy_prop = legacy_props[prop_name]
new_prop = new_props[prop_name]
# For required parameters, should have simple type
if prop_name in legacy_schema.get("required", []):
assert legacy_prop.get("type") == new_prop.get("type")
assert "anyOf" not in legacy_prop
assert "anyOf" not in new_prop
else:
# Both implementations now correctly preserve original schema without nullable behavior
assert "anyOf" not in legacy_prop
assert "anyOf" not in new_prop
# Both should have the same type
assert legacy_prop.get("type") == new_prop.get("type")
async def test_collision_handling_compatibility(self, collision_spec):
"""Test that parameter collision handling is identical."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
# Create both servers
legacy_server = LegacyFastMCPOpenAPI(
openapi_spec=collision_spec,
client=client,
name="Legacy Server",
)
new_server = FastMCPOpenAPI(
openapi_spec=collision_spec,
client=client,
name="New Server",
)
# Get tools from both servers
async with Client(legacy_server) as legacy_client:
legacy_tools = await legacy_client.list_tools()
async with Client(new_server) as new_client:
new_tools = await new_client.list_tools()
# Should have same number of tools
assert len(legacy_tools) == len(new_tools)
assert len(legacy_tools) == 1
# Get the single tool from each
legacy_tool = legacy_tools[0]
new_tool = new_tools[0]
# Input schemas should be identical
legacy_schema = legacy_tool.inputSchema
new_schema = new_tool.inputSchema
# Both should have collision-resolved parameters
legacy_props = legacy_schema.get("properties", {})
new_props = new_schema.get("properties", {})
# Should have: id__path (path param), id (body param), name (body param)
expected_props = {"id__path", "id", "name"}
assert set(legacy_props.keys()) == expected_props
assert set(new_props.keys()) == expected_props
# Required should include path param and required body params
legacy_required = set(legacy_schema.get("required", []))
new_required = set(new_schema.get("required", []))
assert legacy_required == new_required
assert "id__path" in legacy_required
assert "name" in legacy_required
# Path parameter should have integer type
assert legacy_props["id__path"]["type"] == "integer"
assert new_props["id__path"]["type"] == "integer"
# Body parameters should match
assert legacy_props["id"]["type"] == "integer"
assert new_props["id"]["type"] == "integer"
assert legacy_props["name"]["type"] == "string"
assert new_props["name"]["type"] == "string"
async def test_tool_execution_parameter_mapping(self, collision_spec):
"""Test that tool execution with collisions works identically."""
# This test verifies that both implementations can execute the same arguments
# We can't easily test actual HTTP calls, but we can test argument validation
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
# Create both servers
legacy_server = LegacyFastMCPOpenAPI(
openapi_spec=collision_spec,
client=client,
name="Legacy Server",
)
new_server = FastMCPOpenAPI(
openapi_spec=collision_spec,
client=client,
name="New Server",
)
# Test arguments that should work with collision resolution
test_args = {
"id__path": 123, # Path parameter (suffixed)
"id": 456, # Body parameter (not suffixed)
"name": "John Doe", # Body parameter
}
async with Client(legacy_server) as legacy_client:
async with Client(new_server) as new_client:
# Both should accept the same arguments
# We'll test this by attempting to call the tools
# (they'll fail at HTTP level but should pass argument validation)
legacy_tools = await legacy_client.list_tools()
new_tools = await new_client.list_tools()
legacy_tool_name = legacy_tools[0].name
new_tool_name = new_tools[0].name
# Names should be identical
assert legacy_tool_name == new_tool_name
# Both should fail at the HTTP request level (not argument validation)
# This confirms the argument mapping works identically
with pytest.raises(Exception) as legacy_exc:
await legacy_client.call_tool(legacy_tool_name, test_args)
with pytest.raises(Exception) as new_exc:
await new_client.call_tool(new_tool_name, test_args)
# Both should fail with similar error types (HTTP-related, not schema validation)
# The exact error might differ but shouldn't be schema validation errors
legacy_error = str(legacy_exc.value)
new_error = str(new_exc.value)
# Neither should fail due to schema validation
assert "schema" not in legacy_error.lower()
assert "schema" not in new_error.lower()
assert "validation" not in legacy_error.lower()
assert "validation" not in new_error.lower()
async def test_optional_parameter_handling(self, simple_spec):
"""Test that optional parameters are handled identically."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
# Create both servers
legacy_server = LegacyFastMCPOpenAPI(
openapi_spec=simple_spec,
client=client,
name="Legacy Server",
)
new_server = FastMCPOpenAPI(
openapi_spec=simple_spec,
client=client,
name="New Server",
)
# Test with optional parameter omitted (should be None/null)
test_args_minimal = {"id": 123}
# Test with optional parameter included
test_args_full = {"id": 123, "include_details": True}
async with Client(legacy_server) as legacy_client:
async with Client(new_server) as new_client:
legacy_tools = await legacy_client.list_tools()
await new_client.list_tools()
tool_name = legacy_tools[0].name
# Both should handle minimal args the same way
with pytest.raises(Exception) as legacy_exc_min:
await legacy_client.call_tool(tool_name, test_args_minimal)
with pytest.raises(Exception) as new_exc_min:
await new_client.call_tool(tool_name, test_args_minimal)
# Both should handle full args the same way
with pytest.raises(Exception) as legacy_exc_full:
await legacy_client.call_tool(tool_name, test_args_full)
with pytest.raises(Exception) as new_exc_full:
await new_client.call_tool(tool_name, test_args_full)
# All should fail at HTTP level, not schema validation
for exc in [
legacy_exc_min,
new_exc_min,
legacy_exc_full,
new_exc_full,
]:
error_msg = str(exc.value).lower()
assert "schema" not in error_msg
assert "validation" not in error_msg

0
tests/server/openapi/__init__.py → tests/server/providers/openapi/__init__.py
View file

56
tests/server/openapi/test_comprehensive.py → tests/server/providers/openapi/test_comprehensive.py
View file

@ -1,4 +1,4 @@
"""Comprehensive tests for OpenAPI new implementation."""
"""Comprehensive tests for OpenAPIProvider implementation."""
import json
from unittest.mock import AsyncMock, Mock
@ -7,8 +7,21 @@ import httpx
import pytest
from httpx import Response
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.openapi import FastMCPOpenAPI
from fastmcp.server.providers.openapi import OpenAPIProvider
def create_openapi_server(
openapi_spec: dict,
client,
name: str = "OpenAPI Server",
) -> FastMCP:
"""Helper to create a FastMCP server with OpenAPIProvider."""
provider = OpenAPIProvider(openapi_spec=openapi_spec, client=client)
mcp = FastMCP(name)
mcp.add_provider(provider)
return mcp
class TestOpenAPIComprehensive:
@ -357,16 +370,17 @@ class TestOpenAPIComprehensive:
):
"""Test server initialization with comprehensive spec."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
provider = OpenAPIProvider(
openapi_spec=comprehensive_openapi_spec,
client=client,
name="Comprehensive Test Server",
)
server = FastMCP("Comprehensive Test Server")
server.add_provider(provider)
# Should initialize successfully
assert server.name == "Comprehensive Test Server"
assert hasattr(server, "_director")
assert hasattr(server, "_spec")
assert hasattr(provider, "_director")
assert hasattr(provider, "_spec")
# Test with in-memory client
async with Client(server) as mcp_client:
@ -389,7 +403,7 @@ class TestOpenAPIComprehensive:
async def test_openapi_31_compatibility(self, openapi_31_spec):
"""Test that OpenAPI 3.1 specs work correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=openapi_31_spec,
client=client,
name="OpenAPI 3.1 Test",
@ -405,7 +419,7 @@ class TestOpenAPIComprehensive:
async def test_parameter_collision_handling(self, comprehensive_openapi_spec):
"""Test that parameter collisions are handled correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_openapi_spec,
client=client,
)
@ -436,7 +450,7 @@ class TestOpenAPIComprehensive:
async def test_deep_object_parameters(self, comprehensive_openapi_spec):
"""Test deepObject parameter handling."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_openapi_spec,
client=client,
)
@ -480,7 +494,7 @@ class TestOpenAPIComprehensive:
mock_client.send = AsyncMock(return_value=mock_response)
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_openapi_spec,
client=mock_client,
)
@ -517,7 +531,7 @@ class TestOpenAPIComprehensive:
mock_client.send = AsyncMock(return_value=mock_response)
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_openapi_spec,
client=mock_client,
)
@ -561,7 +575,7 @@ class TestOpenAPIComprehensive:
mock_client.send = AsyncMock(return_value=mock_response)
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_openapi_spec,
client=mock_client,
)
@ -608,7 +622,7 @@ class TestOpenAPIComprehensive:
mock_response.raise_for_status = raise_for_status
mock_client.send = AsyncMock(return_value=mock_response)
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_openapi_spec,
client=mock_client,
)
@ -625,7 +639,7 @@ class TestOpenAPIComprehensive:
async def test_schema_refs_resolution(self, comprehensive_openapi_spec):
"""Test that schema references are resolved correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_openapi_spec,
client=client,
)
@ -646,7 +660,7 @@ class TestOpenAPIComprehensive:
async def test_optional_vs_required_parameters(self, comprehensive_openapi_spec):
"""Test handling of optional vs required parameters."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_openapi_spec,
client=client,
)
@ -674,11 +688,11 @@ class TestOpenAPIComprehensive:
"""Test that server initialization is fast (no code generation latency)."""
import time
# Time the server creation
# Time the provider creation
start_time = time.time()
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
provider = OpenAPIProvider(
openapi_spec=comprehensive_openapi_spec,
client=client,
)
@ -689,7 +703,7 @@ class TestOpenAPIComprehensive:
initialization_time = end_time - start_time
assert initialization_time < 0.1 # Should be under 100ms
# Verify server was created correctly
assert server is not None
assert hasattr(server, "_director")
assert hasattr(server, "_spec")
# Verify provider was created correctly
assert provider is not None
assert hasattr(provider, "_director")
assert hasattr(provider, "_spec")

27
tests/server/openapi/test_deepobject_style.py → tests/server/providers/openapi/test_deepobject_style.py
View file

@ -1,10 +1,23 @@
"""Tests for deepObject style parameter handling in openapi_new."""
"""Tests for deepObject style parameter handling in OpenAPIProvider."""
import httpx
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.openapi import FastMCPOpenAPI
from fastmcp.server.providers.openapi import OpenAPIProvider
def create_openapi_server(
openapi_spec: dict,
client,
name: str = "OpenAPI Server",
) -> FastMCP:
"""Helper to create a FastMCP server with OpenAPIProvider."""
provider = OpenAPIProvider(openapi_spec=openapi_spec, client=client)
mcp = FastMCP(name)
mcp.add_provider(provider)
return mcp
class TestDeepObjectStyle:
@ -177,7 +190,7 @@ class TestDeepObjectStyle:
async def test_deepobject_style_parsing_from_spec(self, deepobject_spec):
"""Test that deepObject style parameters are correctly parsed from OpenAPI spec."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=deepobject_spec,
client=client,
name="DeepObject Test Server",
@ -210,7 +223,7 @@ class TestDeepObjectStyle:
async def test_deepobject_explode_true_handling(self, deepobject_spec):
"""Test deepObject with explode=true parameter handling."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=deepobject_spec,
client=client,
name="DeepObject Test Server",
@ -235,7 +248,7 @@ class TestDeepObjectStyle:
async def test_deepobject_explode_false_handling(self, deepobject_spec):
"""Test deepObject with explode=false parameter handling."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=deepobject_spec,
client=client,
name="DeepObject Test Server",
@ -263,7 +276,7 @@ class TestDeepObjectStyle:
async def test_nested_object_structure_in_request_body(self, deepobject_spec):
"""Test nested object structures in request body are preserved."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=deepobject_spec,
client=client,
name="DeepObject Test Server",
@ -307,7 +320,7 @@ class TestDeepObjectStyle:
async def test_deepobject_tool_functionality(self, deepobject_spec):
"""Test that tools with deepObject parameters maintain basic functionality."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=deepobject_spec,
client=client,
name="DeepObject Test Server",

221
tests/server/providers/openapi/test_end_to_end_compatibility.py Normal file
View file

@ -0,0 +1,221 @@
"""End-to-end tests for OpenAPIProvider implementation."""
import httpx
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.providers.openapi import OpenAPIProvider
def create_openapi_server(
openapi_spec: dict,
client,
name: str = "OpenAPI Server",
) -> FastMCP:
"""Helper to create a FastMCP server with OpenAPIProvider."""
provider = OpenAPIProvider(openapi_spec=openapi_spec, client=client)
mcp = FastMCP(name)
mcp.add_provider(provider)
return mcp
class TestEndToEndFunctionality:
"""Test end-to-end functionality of OpenAPIProvider."""
@pytest.fixture
def simple_spec(self):
"""Simple OpenAPI spec for testing."""
return {
"openapi": "3.0.0",
"info": {"title": "Test API", "version": "1.0.0"},
"paths": {
"/users/{id}": {
"get": {
"operationId": "get_user",
"summary": "Get user by ID",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "integer"},
},
{
"name": "include_details",
"in": "query",
"required": False,
"schema": {"type": "boolean"},
},
],
"responses": {"200": {"description": "User found"}},
}
}
},
}
@pytest.fixture
def collision_spec(self):
"""OpenAPI spec with parameter collisions."""
return {
"openapi": "3.0.0",
"info": {"title": "Collision API", "version": "1.0.0"},
"paths": {
"/users/{id}": {
"put": {
"operationId": "update_user",
"summary": "Update user",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "integer"},
}
],
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
},
"required": ["name"],
}
}
},
},
"responses": {"200": {"description": "User updated"}},
}
}
},
}
async def test_tool_schema_generation(self, simple_spec):
"""Test that tools have correct input schemas."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = create_openapi_server(
openapi_spec=simple_spec,
client=client,
name="Test Server",
)
async with Client(server) as mcp_client:
tools = await mcp_client.list_tools()
# Should have one tool
assert len(tools) == 1
tool = tools[0]
assert tool.name == "get_user"
assert tool.description
# Check schema structure
schema = tool.inputSchema
assert schema["type"] == "object"
properties = schema.get("properties", {})
assert "id" in properties
assert "include_details" in properties
# Required fields should include path parameter
required = schema.get("required", [])
assert "id" in required
async def test_collision_handling(self, collision_spec):
"""Test that parameter collision handling works correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = create_openapi_server(
openapi_spec=collision_spec,
client=client,
name="Collision Test Server",
)
async with Client(server) as mcp_client:
tools = await mcp_client.list_tools()
# Should have one tool
assert len(tools) == 1
tool = tools[0]
schema = tool.inputSchema
# Both should have collision-resolved parameters
properties = schema.get("properties", {})
# Should have: id__path (path param), id (body param), name (body param)
expected_props = {"id__path", "id", "name"}
assert set(properties.keys()) == expected_props
# Required should include path param and required body params
required = set(schema.get("required", []))
assert "id__path" in required
assert "name" in required
# Path parameter should have integer type
assert properties["id__path"]["type"] == "integer"
# Body parameters should match
assert properties["id"]["type"] == "integer"
assert properties["name"]["type"] == "string"
async def test_tool_execution_parameter_mapping(self, collision_spec):
"""Test that tool execution with collisions works correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = create_openapi_server(
openapi_spec=collision_spec,
client=client,
name="Test Server",
)
# Test arguments that should work with collision resolution
test_args = {
"id__path": 123, # Path parameter (suffixed)
"id": 456, # Body parameter (not suffixed)
"name": "John Doe", # Body parameter
}
async with Client(server) as mcp_client:
tools = await mcp_client.list_tools()
tool_name = tools[0].name
# Should fail at HTTP level (not argument validation)
# since we don't have an actual server
with pytest.raises(Exception) as exc_info:
await mcp_client.call_tool(tool_name, test_args)
# Should fail at HTTP level, not schema validation
error_msg = str(exc_info.value).lower()
assert "schema" not in error_msg
assert "validation" not in error_msg
async def test_optional_parameter_handling(self, simple_spec):
"""Test that optional parameters are handled correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = create_openapi_server(
openapi_spec=simple_spec,
client=client,
name="Test Server",
)
# Test with optional parameter omitted
test_args_minimal = {"id": 123}
# Test with optional parameter included
test_args_full = {"id": 123, "include_details": True}
async with Client(server) as mcp_client:
tools = await mcp_client.list_tools()
tool_name = tools[0].name
# Both should fail at HTTP level (not argument validation)
for test_args in [test_args_minimal, test_args_full]:
with pytest.raises(Exception) as exc_info:
await mcp_client.call_tool(tool_name, test_args)
error_msg = str(exc_info.value).lower()
assert "schema" not in error_msg
assert "validation" not in error_msg

25
tests/server/openapi/test_openapi_features.py → tests/server/providers/openapi/test_openapi_features.py
View file

@ -1,10 +1,23 @@
"""Tests for OpenAPI feature support in openapi_new."""
"""Tests for OpenAPI feature support in OpenAPIProvider."""
import httpx
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.openapi import FastMCPOpenAPI
from fastmcp.server.providers.openapi import OpenAPIProvider
def create_openapi_server(
openapi_spec: dict,
client,
name: str = "OpenAPI Server",
) -> FastMCP:
"""Helper to create a FastMCP server with OpenAPIProvider."""
provider = OpenAPIProvider(openapi_spec=openapi_spec, client=client)
mcp = FastMCP(name)
mcp.add_provider(provider)
return mcp
class TestParameterHandling:
@ -127,7 +140,7 @@ class TestParameterHandling:
async def test_query_parameters_in_tools(self, parameter_spec):
"""Test that query parameters are properly included in tool parameters."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=parameter_spec, client=client, name="Parameter Test Server"
)
@ -175,7 +188,7 @@ class TestParameterHandling:
async def test_path_parameters_in_tools(self, parameter_spec):
"""Test that path parameters are properly included in tool parameters."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=parameter_spec, client=client, name="Parameter Test Server"
)
@ -280,7 +293,7 @@ class TestRequestBodyHandling:
async def test_request_body_properties_in_tool(self, request_body_spec):
"""Test that request body properties are included in tool parameters."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=request_body_spec,
client=client,
name="Request Body Test Server",
@ -381,7 +394,7 @@ class TestResponseSchemas:
async def test_tool_has_output_schema(self, response_schema_spec):
"""Test that tools have output schemas from response definitions."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=response_schema_spec,
client=client,
name="Response Schema Test Server",

0
tests/server/openapi/test_openapi_performance.py → tests/server/providers/openapi/test_openapi_performance.py
View file

23
tests/server/openapi/test_parameter_collisions.py → tests/server/providers/openapi/test_parameter_collisions.py
View file

@ -1,10 +1,23 @@
"""Tests for parameter collision handling in openapi_new."""
"""Tests for parameter collision handling in OpenAPIProvider."""
import httpx
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.openapi import FastMCPOpenAPI
from fastmcp.server.providers.openapi import OpenAPIProvider
def create_openapi_server(
openapi_spec: dict,
client,
name: str = "OpenAPI Server",
) -> FastMCP:
"""Helper to create a FastMCP server with OpenAPIProvider."""
provider = OpenAPIProvider(openapi_spec=openapi_spec, client=client)
mcp = FastMCP(name)
mcp.add_provider(provider)
return mcp
class TestParameterCollisions:
@ -121,7 +134,7 @@ class TestParameterCollisions:
async def test_path_body_collision_handling(self, collision_spec):
"""Test that path and body parameters with same name are handled correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=collision_spec, client=client, name="Collision Test Server"
)
@ -161,7 +174,7 @@ class TestParameterCollisions:
async def test_query_header_collision_handling(self, collision_spec):
"""Test that query and header parameters with same name are handled correctly."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=collision_spec, client=client, name="Collision Test Server"
)
@ -191,7 +204,7 @@ class TestParameterCollisions:
async def test_collision_resolution_maintains_functionality(self, collision_spec):
"""Test that collision resolution doesn't break basic tool functionality."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=collision_spec, client=client, name="Collision Test Server"
)

158
tests/server/openapi/test_performance_comparison.py → tests/server/providers/openapi/test_performance_comparison.py
View file

@ -1,16 +1,29 @@
"""Performance comparison between legacy and new OpenAPI implementations."""
"""Performance tests for OpenAPIProvider implementation."""
import gc
import time
import httpx
import pytest
from fastmcp.server.openapi import FastMCPOpenAPI
from fastmcp.server.openapi import FastMCPOpenAPI as LegacyFastMCPOpenAPI
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
class TestPerformanceComparison:
"""Compare performance between legacy and new implementations."""
def create_openapi_server(
openapi_spec: dict,
client,
name: str = "OpenAPI Server",
) -> FastMCP:
"""Helper to create a FastMCP server with OpenAPIProvider."""
provider = OpenAPIProvider(openapi_spec=openapi_spec, client=client)
mcp = FastMCP(name)
mcp.add_provider(provider)
return mcp
class TestPerformance:
"""Test performance of OpenAPIProvider implementation."""
@pytest.fixture
def comprehensive_spec(self):
@ -153,94 +166,83 @@ class TestPerformanceComparison:
},
}
def test_server_initialization_performance(self, comprehensive_spec):
"""Test that new implementation is significantly faster than legacy."""
def test_provider_initialization_performance(self, comprehensive_spec):
"""Test that provider initialization is fast (serverless requirement)."""
num_iterations = 5
# Measure legacy implementation
legacy_times = []
# Measure provider initialization
times = []
for _ in range(num_iterations):
client = httpx.AsyncClient(base_url="https://api.example.com")
start_time = time.time()
server = LegacyFastMCPOpenAPI(
provider = OpenAPIProvider(
openapi_spec=comprehensive_spec,
client=client,
name="Legacy Performance Test",
)
# Ensure server is fully initialized
assert server is not None
# Ensure provider is fully initialized
assert provider is not None
end_time = time.time()
legacy_times.append(end_time - start_time)
times.append(end_time - start_time)
# Measure new implementation
new_times = []
for _ in range(num_iterations):
client = httpx.AsyncClient(base_url="https://api.example.com")
start_time = time.time()
server = FastMCPOpenAPI(
openapi_spec=comprehensive_spec,
client=client,
name="New Performance Test",
)
# Ensure server is fully initialized
assert server is not None
end_time = time.time()
new_times.append(end_time - start_time)
# Calculate averages
legacy_avg = sum(legacy_times) / len(legacy_times)
new_avg = sum(new_times) / len(new_times)
print(f"Legacy implementation average: {legacy_avg:.4f}s")
print(f"New implementation average: {new_avg:.4f}s")
print(f"Speedup: {legacy_avg / new_avg:.2f}x")
# Both implementations should be very fast for moderate specs
# The key achievement is eliminating the 100-200ms latency issue for serverless
avg_time = sum(times) / len(times)
max_acceptable_time = 0.1 # 100ms
print(f"Legacy performance: {'✓' if legacy_avg < max_acceptable_time else '✗'}")
print(f"New performance: {'✓' if new_avg < max_acceptable_time else '✗'}")
print(f"Average initialization time: {avg_time:.4f}s")
print(f"Performance: {'✓' if avg_time < max_acceptable_time else '✗'}")
# New implementation should be under 100ms for reasonable specs (serverless requirement)
assert new_avg < max_acceptable_time, (
f"New implementation should initialize in under 100ms, got {new_avg:.4f}s"
# Should initialize in under 100ms for serverless requirements
assert avg_time < max_acceptable_time, (
f"Provider should initialize in under 100ms, got {avg_time:.4f}s"
)
# Legacy might be slightly faster or slower on small specs, but both should be fast
# The real improvement shows up with larger specs where code generation was the bottleneck
assert legacy_avg < max_acceptable_time, (
f"Legacy should also be fast on small specs, got {legacy_avg:.4f}s"
def test_server_initialization_performance(self, comprehensive_spec):
"""Test that full server initialization is fast."""
num_iterations = 5
times = []
for _ in range(num_iterations):
client = httpx.AsyncClient(base_url="https://api.example.com")
start_time = time.time()
server = create_openapi_server(
openapi_spec=comprehensive_spec,
client=client,
name="Performance Test",
)
# Ensure server is fully initialized
assert server is not None
end_time = time.time()
times.append(end_time - start_time)
avg_time = sum(times) / len(times)
max_acceptable_time = 0.1 # 100ms
print(f"Average server initialization time: {avg_time:.4f}s")
assert avg_time < max_acceptable_time, (
f"Server should initialize in under 100ms, got {avg_time:.4f}s"
)
def test_functionality_identical_after_optimization(self, comprehensive_spec):
def test_functionality_after_optimization(self, comprehensive_spec):
"""Verify that performance optimization doesn't break functionality."""
client = httpx.AsyncClient(base_url="https://api.example.com")
# Create both servers
legacy_server = LegacyFastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_spec,
client=client,
name="Legacy Server",
)
new_server = FastMCPOpenAPI(
openapi_spec=comprehensive_spec,
client=client,
name="New Server",
name="Test Server",
)
# Both should have the same number of tools
legacy_tool_count = len(legacy_server._tool_manager._tools)
new_tool_count = len(new_server._tool_manager._tools)
# Get tools from the provider
def get_provider_tools(server):
for provider in server._providers:
if hasattr(provider, "_tools"):
return provider._tools
return {}
assert legacy_tool_count == new_tool_count
assert legacy_tool_count == 6 # 6 operations in the spec
tools = get_provider_tools(server)
# Tool names should be identical
legacy_tool_names = set(legacy_server._tool_manager._tools.keys())
new_tool_names = set(new_server._tool_manager._tools.keys())
assert legacy_tool_names == new_tool_names
# Should have 6 operations in the spec
assert len(tools) == 6
# Expected operations
expected_operations = {
@ -251,20 +253,25 @@ class TestPerformanceComparison:
"delete_user",
"search_users",
}
assert legacy_tool_names == expected_operations
assert set(tools.keys()) == expected_operations
def test_memory_efficiency(self, comprehensive_spec):
"""Test that new implementation doesn't significantly increase memory usage."""
import gc
"""Test that implementation doesn't significantly increase memory usage."""
# Helper to get tools from provider
def get_provider_tools(server):
for provider in server._providers:
if hasattr(provider, "_tools"):
return provider._tools
return {}
# This is a basic test - in practice you'd use more sophisticated memory profiling
gc.collect() # Clean up before baseline
baseline_refs = len(gc.get_objects())
servers = []
for i in range(10):
client = httpx.AsyncClient(base_url="https://api.example.com")
server = FastMCPOpenAPI(
server = create_openapi_server(
openapi_spec=comprehensive_spec,
client=client,
name=f"Memory Test Server {i}",
@ -273,12 +280,11 @@ class TestPerformanceComparison:
# Servers should all be functional
assert len(servers) == 10
assert all(len(s._tool_manager._tools) == 6 for s in servers)
assert all(len(get_provider_tools(s)) == 6 for s in servers)
# Memory usage shouldn't explode (this is a basic check)
gc.collect() # Clean up
# Memory usage shouldn't explode
gc.collect()
current_refs = len(gc.get_objects())
# Allow reasonable memory growth but not exponential
growth_ratio = current_refs / max(baseline_refs, 1)
assert growth_ratio < 3.0, (
f"Memory usage grew by {growth_ratio}x, which seems excessive"

122
tests/server/openapi/test_server.py → tests/server/providers/openapi/test_server.py
View file

@ -1,14 +1,15 @@
"""Unit tests for FastMCPOpenAPI server."""
"""Unit tests for OpenAPIProvider."""
import httpx
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.openapi import FastMCPOpenAPI
from fastmcp.server.providers.openapi import OpenAPIProvider
class TestFastMCPOpenAPIBasicFunctionality:
"""Test basic FastMCPOpenAPI server functionality."""
class TestOpenAPIProviderBasicFunctionality:
"""Test basic OpenAPIProvider functionality."""
@pytest.fixture
def simple_openapi_spec(self):
@ -90,37 +91,34 @@ class TestFastMCPOpenAPIBasicFunctionality:
},
}
def test_server_initialization(self, simple_openapi_spec):
"""Test server initialization with OpenAPI spec."""
def test_provider_initialization(self, simple_openapi_spec):
"""Test provider initialization with OpenAPI spec."""
client = httpx.AsyncClient(base_url="https://api.example.com")
provider = OpenAPIProvider(openapi_spec=simple_openapi_spec, client=client)
server = FastMCPOpenAPI(
openapi_spec=simple_openapi_spec, client=client, name="Test Server"
)
assert server.name == "Test Server"
# Should have initialized RequestDirector successfully
assert hasattr(server, "_director")
assert hasattr(server, "_spec")
assert hasattr(provider, "_director")
assert hasattr(provider, "_spec")
def test_server_initialization_with_custom_name(self, simple_openapi_spec):
"""Test server initialization with custom name."""
def test_server_with_provider(self, simple_openapi_spec):
"""Test server initialization with OpenAPIProvider."""
client = httpx.AsyncClient(base_url="https://api.example.com")
provider = OpenAPIProvider(openapi_spec=simple_openapi_spec, client=client)
server = FastMCPOpenAPI(openapi_spec=simple_openapi_spec, client=client)
mcp = FastMCP("Test Server")
mcp.add_provider(provider)
# Should use default name
assert server.name == "OpenAPI FastMCP"
assert mcp.name == "Test Server"
async def test_server_creates_tools_from_spec(self, simple_openapi_spec):
"""Test that server creates tools from OpenAPI spec."""
async def test_provider_creates_tools_from_spec(self, simple_openapi_spec):
"""Test that provider creates tools from OpenAPI spec."""
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
openapi_spec=simple_openapi_spec, client=client, name="Test Server"
)
provider = OpenAPIProvider(openapi_spec=simple_openapi_spec, client=client)
# Test with in-memory client
async with Client(server) as mcp_client:
mcp = FastMCP("Test Server")
mcp.add_provider(provider)
async with Client(mcp) as mcp_client:
tools = await mcp_client.list_tools()
# Should have created tools for both operations
@ -130,60 +128,37 @@ class TestFastMCPOpenAPIBasicFunctionality:
assert "get_user" in tool_names
assert "create_user" in tool_names
async def test_server_tool_execution_fallback_to_http(self, simple_openapi_spec):
"""Test tool execution falls back to HTTP when callables aren't available."""
# Use a mock client that will be used for HTTP fallback
async def test_provider_tool_execution(self, simple_openapi_spec):
"""Test tool execution uses RequestDirector."""
mock_client = httpx.AsyncClient()
provider = OpenAPIProvider(openapi_spec=simple_openapi_spec, client=mock_client)
server = FastMCPOpenAPI(
openapi_spec=simple_openapi_spec, client=mock_client, name="Test Server"
)
mcp = FastMCP("Test Server")
mcp.add_provider(provider)
# With new architecture, tools are always created using RequestDirector
async with Client(server) as mcp_client:
async with Client(mcp) as mcp_client:
tools = await mcp_client.list_tools()
# Should still have tools even without callables
# Should have tools using RequestDirector
assert len(tools) == 2
# Tools should be OpenAPITool instances using RequestDirector
# We'll just verify they exist and are callable
get_user_tool = next(tool for tool in tools if tool.name == "get_user")
assert get_user_tool is not None
assert get_user_tool.description is not None
def test_server_request_director_initialization(self, simple_openapi_spec):
"""Test that server initializes RequestDirector successfully."""
def test_provider_with_timeout(self, simple_openapi_spec):
"""Test provider initialization with timeout setting."""
client = httpx.AsyncClient(base_url="https://api.example.com")
# This should not raise an exception
server = FastMCPOpenAPI(
openapi_spec=simple_openapi_spec, client=client, name="Test Server"
)
# Server should be created successfully
assert server is not None
assert server.name == "Test Server"
# RequestDirector and Spec should be initialized
assert hasattr(server, "_director")
assert hasattr(server, "_spec")
def test_server_with_timeout(self, simple_openapi_spec):
"""Test server initialization with timeout setting."""
client = httpx.AsyncClient(base_url="https://api.example.com")
server = FastMCPOpenAPI(
provider = OpenAPIProvider(
openapi_spec=simple_openapi_spec,
client=client,
name="Test Server",
timeout=30.0,
)
assert server._timeout == 30.0
assert provider._timeout == 30.0
def test_server_with_empty_spec(self):
"""Test server with minimal OpenAPI spec."""
def test_provider_with_empty_spec(self):
"""Test provider with minimal OpenAPI spec."""
minimal_spec = {
"openapi": "3.0.0",
"info": {"title": "Empty API", "version": "1.0.0"},
@ -191,19 +166,14 @@ class TestFastMCPOpenAPIBasicFunctionality:
}
client = httpx.AsyncClient(base_url="https://api.example.com")
provider = OpenAPIProvider(openapi_spec=minimal_spec, client=client)
server = FastMCPOpenAPI(
openapi_spec=minimal_spec, client=client, name="Empty Server"
)
assert server.name == "Empty Server"
# Should handle empty paths gracefully
assert hasattr(server, "_director")
assert hasattr(server, "_spec")
assert hasattr(provider, "_director")
assert hasattr(provider, "_spec")
async def test_clean_schema_output_no_unused_defs(self):
"""Test that unused schema definitions are removed from tool schemas."""
# Create a spec with unused HTTPValidationError-like definitions
spec_with_unused_defs = {
"openapi": "3.0.0",
"info": {"title": "Test API", "version": "1.0.0"},
@ -264,7 +234,6 @@ class TestFastMCPOpenAPIBasicFunctionality:
},
"components": {
"schemas": {
# This should be removed since it's not referenced
"HTTPValidationError": {
"properties": {
"detail": {
@ -299,20 +268,20 @@ class TestFastMCPOpenAPIBasicFunctionality:
}
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
server = FastMCPOpenAPI(
openapi_spec=spec_with_unused_defs, client=client, name="Test Server"
provider = OpenAPIProvider(
openapi_spec=spec_with_unused_defs, client=client
)
mcp = FastMCP("Test Server")
mcp.add_provider(provider)
async with Client(server) as mcp_client:
async with Client(mcp) as mcp_client:
tools = await mcp_client.list_tools()
assert len(tools) == 1 # Only the POST operation
assert len(tools) == 1
tool = tools[0]
# Verify tool has clean schemas without unused $defs
assert tool.name == "create_user"
# Input schema should not have $defs since no references are used
expected_input_schema = {
"type": "object",
"properties": {
@ -323,7 +292,6 @@ class TestFastMCPOpenAPIBasicFunctionality:
}
assert tool.inputSchema == expected_input_schema
# Output schema should not have $defs since no references are used
expected_output_schema = {
"type": "object",
"properties": {