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

Convert OpenAPI provider to the OpenAPI plugin (#4015)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-04-22 13:56:56 -04:00 • committed by GitHub
commit a82979e433
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 1700 additions and 1021 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,4 +1,4 @@
"""Deprecated: Import from fastmcp.server.providers.openapi instead."""
"""Deprecated: Import from fastmcp.server.plugins.openapi instead."""
import warnings
@ -7,23 +7,27 @@ from fastmcp.exceptions import FastMCPDeprecationWarning
# 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.providers.openapi instead.",
"Import from fastmcp.server.plugins.openapi instead.",
FastMCPDeprecationWarning,
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,
from fastmcp.server.plugins.openapi import ( # noqa: E402
MCPType as MCPType,
RouteMap as RouteMap,
)
from fastmcp.server.plugins.openapi.components import ( # noqa: E402
OpenAPIResource as OpenAPIResource,
OpenAPIResourceTemplate as OpenAPIResourceTemplate,
OpenAPITool as OpenAPITool,
RouteMap as RouteMap,
)
from fastmcp.server.plugins.openapi.routing import ( # noqa: E402
ComponentFn as ComponentFn,
RouteMapFn as RouteMapFn,
)
from fastmcp.server.providers.openapi.routing import ( # noqa: E402
from fastmcp.server.plugins.openapi.routing import ( # noqa: E402
DEFAULT_ROUTE_MAPPINGS as DEFAULT_ROUTE_MAPPINGS,
_determine_route_type as _determine_route_type,
)

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

@ -1,12 +1,12 @@
"""OpenAPI server implementation for FastMCP.
.. deprecated::
This module is deprecated. Import from fastmcp.server.providers.openapi instead.
This module is deprecated. Import from fastmcp.server.plugins.openapi instead.
The recommended approach is to use OpenAPIProvider with FastMCP:
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi import OpenAPIProvider
import httpx
client = httpx.AsyncClient(base_url="https://api.example.com")
@ -24,20 +24,26 @@ from fastmcp.exceptions import FastMCPDeprecationWarning
warnings.warn(
"fastmcp.server.openapi is deprecated. "
"Import from fastmcp.server.providers.openapi instead.",
"Import from fastmcp.server.plugins.openapi instead.",
FastMCPDeprecationWarning,
stacklevel=2,
)
# Re-export from new canonical location
from fastmcp.server.providers.openapi import ( # noqa: E402
ComponentFn as ComponentFn,
from fastmcp.server.plugins.openapi import ( # noqa: E402
MCPType as MCPType,
OpenAPIProvider as OpenAPIProvider,
RouteMap as RouteMap,
)
from fastmcp.server.plugins.openapi.components import ( # noqa: E402
OpenAPIResource as OpenAPIResource,
OpenAPIResourceTemplate as OpenAPIResourceTemplate,
OpenAPITool as OpenAPITool,
RouteMap as RouteMap,
)
from fastmcp.server.plugins.openapi.provider import ( # noqa: E402
OpenAPIProvider as OpenAPIProvider,
)
from fastmcp.server.plugins.openapi.routing import ( # noqa: E402
ComponentFn as ComponentFn,
RouteMapFn as RouteMapFn,
)

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

@ -1,6 +1,6 @@
"""OpenAPI component implementations - backwards compatibility stub.
This module is deprecated. Import from fastmcp.server.providers.openapi instead.
This module is deprecated. Import from fastmcp.server.plugins.openapi instead.
"""
from __future__ import annotations
@ -11,12 +11,12 @@ from fastmcp.exceptions import FastMCPDeprecationWarning
warnings.warn(
"fastmcp.server.openapi.components is deprecated. "
"Import from fastmcp.server.providers.openapi instead.",
"Import from fastmcp.server.plugins.openapi instead.",
FastMCPDeprecationWarning,
stacklevel=2,
)
from fastmcp.server.providers.openapi import ( # noqa: E402
from fastmcp.server.plugins.openapi.components import ( # noqa: E402
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,

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

@ -22,27 +22,27 @@ __all__ = [
warnings.warn(
"fastmcp.server.openapi.routing is deprecated. "
"Import from fastmcp.server.providers.openapi instead.",
"Import from fastmcp.server.plugins.openapi instead.",
FastMCPDeprecationWarning,
stacklevel=2,
)
# Re-export from new canonical location
from fastmcp.server.providers.openapi.routing import (
from fastmcp.server.plugins.openapi.routing import (
DEFAULT_ROUTE_MAPPINGS as DEFAULT_ROUTE_MAPPINGS,
)
from fastmcp.server.providers.openapi.routing import (
from fastmcp.server.plugins.openapi.routing import (
ComponentFn as ComponentFn,
)
from fastmcp.server.providers.openapi.routing import (
from fastmcp.server.plugins.openapi.routing import (
MCPType as MCPType,
)
from fastmcp.server.providers.openapi.routing import (
from fastmcp.server.plugins.openapi.routing import (
RouteMap as RouteMap,
)
from fastmcp.server.providers.openapi.routing import (
from fastmcp.server.plugins.openapi.routing import (
RouteMapFn as RouteMapFn,
)
from fastmcp.server.providers.openapi.routing import (
from fastmcp.server.plugins.openapi.routing import (
_determine_route_type as _determine_route_type,
)

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

@ -3,7 +3,7 @@
This class is deprecated. Use FastMCP with OpenAPIProvider instead:
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi import OpenAPIProvider
import httpx
client = httpx.AsyncClient(base_url="https://api.example.com")
@ -19,12 +19,9 @@ from typing import Any
import httpx
from fastmcp.exceptions import FastMCPDeprecationWarning
from fastmcp.server.providers.openapi import (
ComponentFn,
OpenAPIProvider,
RouteMap,
RouteMapFn,
)
from fastmcp.server.plugins.openapi import RouteMap
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
from fastmcp.server.plugins.openapi.routing import ComponentFn, RouteMapFn
from fastmcp.server.server import FastMCP
@ -49,7 +46,7 @@ class FastMCPOpenAPI(FastMCP):
New approach:
```python
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi import OpenAPIProvider
import httpx
client = httpx.AsyncClient(base_url="https://api.example.com")

21
src/fastmcp/server/plugins/openapi/__init__.py Normal file
View file

@ -0,0 +1,21 @@
"""OpenAPI plugin — mount an OpenAPI spec as MCP tools/resources.
from fastmcp import FastMCP
from fastmcp.server.plugins.openapi import OpenAPI, OpenAPIConfig
mcp = FastMCP(
"Petstore",
plugins=[OpenAPI(OpenAPIConfig(spec=petstore_spec))],
)
Typed `RouteMap` + `MCPType` are re-exported for the Python-only
escape hatch on `OpenAPI.__init__(route_maps=...)`. Everything else
(component classes, provider class, callable type aliases) lives on the
submodules — import from `.provider`, `.components`, `.routing` directly
if you need them.
"""
from fastmcp.server.plugins.openapi.plugin import OpenAPI, OpenAPIConfig
from fastmcp.server.plugins.openapi.routing import MCPType, RouteMap
__all__ = ["MCPType", "OpenAPI", "OpenAPIConfig", "RouteMap"]

421
src/fastmcp/server/plugins/openapi/components.py Normal file
View file

@ -0,0 +1,421 @@
"""OpenAPI component classes: Tool, Resource, and ResourceTemplate."""
from __future__ import annotations
import json
import re
import warnings
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
import httpx
from mcp.types import ToolAnnotations
from pydantic.networks import AnyUrl
import fastmcp
from fastmcp.exceptions import FastMCPDeprecationWarning
from fastmcp.resources import (
Resource,
ResourceContent,
ResourceResult,
ResourceTemplate,
)
from fastmcp.server.dependencies import get_http_headers
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.tools.base 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
_SAFE_HEADERS = frozenset(
{
"accept",
"accept-encoding",
"accept-language",
"cache-control",
"connection",
"content-length",
"content-type",
"host",
"user-agent",
}
)
def _redact_headers(headers: httpx.Headers) -> dict[str, str]:
return {k: v if k.lower() in _SAFE_HEADERS else "***" for k, v in headers.items()}
__all__ = [
"OpenAPIResource",
"OpenAPIResourceTemplate",
"OpenAPITool",
"_extract_mime_type_from_route",
]
logger = get_logger(__name__)
# Default MIME type when no response content type can be inferred
_DEFAULT_MIME_TYPE = "application/json"
def _extract_mime_type_from_route(route: HTTPRoute) -> str:
"""Extract the primary MIME type from an HTTPRoute's response definitions.
Looks for the first successful response (2xx) and returns its content type.
Prefers JSON-compatible types when multiple are available.
Falls back to "application/json" when no response content type is declared.
"""
if not route.responses:
return _DEFAULT_MIME_TYPE
# Priority order for success status codes
success_codes = ["200", "201", "202", "204"]
response_info = None
for status_code in success_codes:
if status_code in route.responses:
response_info = route.responses[status_code]
break
# If no explicit success codes, try any 2xx response
if response_info is None:
for status_code, resp_info in route.responses.items():
if status_code.startswith("2"):
response_info = resp_info
break
if response_info is None or not response_info.content_schema:
return _DEFAULT_MIME_TYPE
# If there's only one content type, use it directly
content_types = list(response_info.content_schema.keys())
if len(content_types) == 1:
return content_types[0]
# When multiple types exist, prefer JSON-compatible types
json_compatible_types = [
"application/json",
"application/vnd.api+json",
"application/hal+json",
"application/ld+json",
"text/json",
]
for ct in json_compatible_types:
if ct in response_info.content_schema:
return ct
# Fall back to the first available content type
return content_types[0]
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,
annotations: ToolAnnotations | None = None,
serializer: Callable[[Any], str] | None = None, # Deprecated
):
if serializer is not None and fastmcp.settings.deprecation_warnings:
warnings.warn(
"The `serializer` parameter is deprecated. "
"Return ToolResult from your tools for full control over serialization. "
"See https://gofastmcp.com/servers/tools#custom-serialization for migration examples.",
FastMCPDeprecationWarning,
stacklevel=2,
)
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
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."""
# Build the request — errors here are programming/schema issues,
# not HTTP failures, so we catch them separately.
try:
base_url = str(self._client.base_url) or "http://localhost"
request = self._director.build(self._route, arguments, base_url)
if self._client.headers:
for key, value in self._client.headers.items():
if key not in request.headers:
request.headers[key] = value
mcp_headers = get_http_headers()
if mcp_headers:
for key, value in mcp_headers.items():
if key not in request.headers:
request.headers[key] = value
except Exception as e:
raise ValueError(
f"Error building request for {self._route.method.upper()} "
f"{self._route.path}: {type(e).__name__}: {e}"
) from e
# Send the request and process the response.
try:
logger.debug(
f"run - sending request; headers: {_redact_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
# Structured content must be a dict for the MCP protocol.
# Wrap non-dict values that slipped through (e.g. a backend
# returning an array when the schema declared an object).
if not isinstance(structured_output, dict):
structured_output = {"result": structured_output}
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.TimeoutException as e:
raise ValueError(f"HTTP request timed out ({type(e).__name__})") from e
except httpx.RequestError as e:
raise ValueError(f"Request error ({type(e).__name__}): {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,
):
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
def __repr__(self) -> str:
return f"OpenAPIResource(name={self.name!r}, uri={self.uri!r}, path={self._route.path})"
async def read(self) -> ResourceResult:
"""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,
)
response.raise_for_status()
content_type = response.headers.get("content-type", "").lower()
if "application/json" in content_type:
result = response.json()
return ResourceResult(
contents=[
ResourceContent(
content=json.dumps(result), mime_type="application/json"
)
]
)
elif any(ct in content_type for ct in ["text/", "application/xml"]):
return ResourceResult(
contents=[
ResourceContent(content=response.text, mime_type=self.mime_type)
]
)
else:
return ResourceResult(
contents=[
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.TimeoutException as e:
raise ValueError(f"HTTP request timed out ({type(e).__name__})") from e
except httpx.RequestError as e:
raise ValueError(f"Request error ({type(e).__name__}): {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,
mime_type: str = _DEFAULT_MIME_TYPE,
):
super().__init__(
uri_template=uri_template,
name=name,
description=description,
parameters=parameters,
tags=tags or set(),
mime_type=mime_type,
)
self._client = client
self._route = route
self._director = director
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=self.mime_type,
tags=set(self._route.tags or []),
)

247
src/fastmcp/server/plugins/openapi/plugin.py Normal file
View file

@ -0,0 +1,247 @@
"""OpenAPI plugin: wrap an OpenAPI spec into an MCP server via the
`OpenAPIProvider`.
The plugin is the JSON-configurable entry point for the OpenAPI
integration. Spec, base URL, headers, timeout, and route mappings can
all be declared in a plugin config (useful for `plugins.json`, Horizon
config forms, or anywhere else you want to spin up an OpenAPI server
without writing Python). For scenarios that need a custom
`httpx.AsyncClient` or callables (`route_map_fn`, `mcp_component_fn`),
pass them through `__init__` directly.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Literal
import httpx
from pydantic import BaseModel, ConfigDict
from fastmcp.server.plugins.base import Plugin, PluginMeta
from fastmcp.server.plugins.openapi.provider import (
OpenAPIProvider,
resolve_spec_base_url,
)
from fastmcp.server.plugins.openapi.routing import (
ComponentFn,
MCPType,
RouteMap,
RouteMapFn,
)
from fastmcp.server.providers import Provider
from fastmcp.utilities.openapi.models import HttpMethod
class RouteMapDict(BaseModel):
"""JSON-serializable form of `RouteMap`.
Converted to a real `RouteMap` when the plugin builds the provider.
The `pattern` field is always a regex string (the typed `RouteMap`
accepts a compiled `Pattern` too, but Config stays JSON-friendly).
"""
model_config = ConfigDict(extra="forbid")
mcp_type: Literal["TOOL", "RESOURCE", "RESOURCE_TEMPLATE", "EXCLUDE"]
"""Target component type. Matches `MCPType` enum values."""
methods: list[HttpMethod] | Literal["*"] = "*"
"""HTTP methods to match (e.g. `["GET", "POST"]`) or `"*"` for any."""
pattern: str = r".*"
"""Regex pattern matched against the route path."""
tags: list[str] = []
"""Route tags that must all be present for this mapping to apply."""
mcp_tags: list[str] = []
"""Tags to attach to the generated MCP component."""
def to_route_map(self) -> RouteMap:
methods: list[HttpMethod] | Literal["*"] = (
"*" if self.methods == "*" else list(self.methods)
)
return RouteMap(
methods=methods,
pattern=self.pattern,
tags=set(self.tags),
mcp_type=MCPType[self.mcp_type],
mcp_tags=set(self.mcp_tags),
)
class OpenAPIConfig(BaseModel):
"""Config model for the `OpenAPI` plugin.
Exactly one of `spec` or `spec_path` must be set — the check fires
when the plugin builds its provider, not at Config construction,
so that `OpenAPIConfig()` with no args still satisfies the
plugin-framework's defaults-are-instantiable contract.
For specs that need to be fetched from a URL at startup, fetch the
dict in your application code and pass it via `spec=...`.
"""
model_config = ConfigDict(extra="forbid")
spec: dict[str, Any] | None = None
"""Inline OpenAPI spec as a dict."""
spec_path: str | None = None
"""Path to a local JSON file containing the OpenAPI spec."""
base_url: str | None = None
"""Base URL for the default httpx client. If omitted, the first
server URL from the spec is used."""
headers: dict[str, str] | None = None
"""Default headers added to every request the generated client
makes."""
timeout_secs: float = 30.0
"""Default timeout (seconds) for the generated httpx client."""
mcp_names: dict[str, str] | None = None
"""Mapping from OpenAPI `operationId` to the MCP component name
that gets generated for it."""
tags: list[str] = []
"""Tags applied to every generated MCP component."""
validate_output: bool = True
"""When true (default), generated tools use the OpenAPI response
schema for output validation. Set false to accept any shape."""
route_maps: list[RouteMapDict] = []
"""Ordered route-mapping rules. First match wins. If omitted, all
routes become tools."""
class OpenAPI(Plugin[OpenAPIConfig]):
"""Mount an OpenAPI spec as an MCP server via a plugin.
Everything declarative (spec, base URL, headers, route mappings)
goes in `OpenAPIConfig`. Python-only knobs — custom `httpx.AsyncClient`,
route-mapping callables, component customization — go in `__init__`
kwargs.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.plugins.openapi import OpenAPI, OpenAPIConfig
# Declarative (JSON-friendly):
mcp = FastMCP(
"Petstore",
plugins=[
OpenAPI(
OpenAPIConfig(
spec=petstore_spec,
base_url="https://api.example.com",
headers={"Authorization": "Bearer ..."},
)
)
],
)
# With a custom httpx client (shared auth, retries, etc.):
custom_client = httpx.AsyncClient(...)
mcp = FastMCP(
"Petstore",
plugins=[
OpenAPI(
OpenAPIConfig(spec=petstore_spec),
client=custom_client,
)
],
)
```
"""
# "OpenAPI" is a single technical term; the auto-kebab would split
# it into "open-api", which is uglier than the established spelling.
meta = PluginMeta(name="openapi")
def __init__(
self,
config: OpenAPIConfig | dict[str, Any] | None = None,
*,
client: httpx.AsyncClient | None = None,
route_maps: list[RouteMap] | None = None,
route_map_fn: RouteMapFn | None = None,
mcp_component_fn: ComponentFn | None = None,
) -> None:
super().__init__(config)
self._client_override = client
self._route_maps_override = route_maps
self._route_map_fn = route_map_fn
self._mcp_component_fn = mcp_component_fn
def providers(self) -> list[Provider]:
spec = self._load_spec()
if self._client_override is not None:
client = self._client_override
# User-supplied client: they own the lifecycle.
owns_client: bool | None = None
else:
client = self._build_default_client(spec)
# Plugin built the client, so the provider lifespan must
# close it on shutdown (default ownership heuristic would
# miss this since `client` is not None by the time we pass
# it in).
owns_client = True
route_maps = self._resolve_route_maps()
return [
OpenAPIProvider(
openapi_spec=spec,
client=client,
route_maps=route_maps,
route_map_fn=self._route_map_fn,
mcp_component_fn=self._mcp_component_fn,
mcp_names=self.config.mcp_names,
tags=set(self.config.tags) if self.config.tags else None,
validate_output=self.config.validate_output,
_owns_client=owns_client,
)
]
def _load_spec(self) -> dict[str, Any]:
if self.config.spec is not None and self.config.spec_path is not None:
raise ValueError(
"OpenAPIConfig requires exactly one of `spec` or `spec_path`, not both."
)
if self.config.spec is not None:
return self.config.spec
if self.config.spec_path is not None:
# Force UTF-8 rather than relying on the process locale —
# OpenAPI specs can carry non-ASCII descriptions and we want
# cross-platform (e.g. Windows cp1252) loads to work.
return json.loads(Path(self.config.spec_path).read_text(encoding="utf-8"))
raise ValueError(
"OpenAPIConfig requires `spec` (inline dict) or `spec_path` "
"(local JSON file) to be set."
)
def _build_default_client(self, spec: dict[str, Any]) -> httpx.AsyncClient:
kwargs: dict[str, Any] = {
"base_url": self.config.base_url or resolve_spec_base_url(spec),
"timeout": self.config.timeout_secs,
}
if self.config.headers:
kwargs["headers"] = self.config.headers
return httpx.AsyncClient(**kwargs)
def _resolve_route_maps(self) -> list[RouteMap] | None:
# Typed override wins over dict-form config so power users who
# pass real RouteMap objects aren't shadowed by an empty default.
if self._route_maps_override is not None:
return self._route_maps_override
if self.config.route_maps:
return [rm.to_route_map() for rm in self.config.route_maps]
return None
__all__ = ["OpenAPI", "OpenAPIConfig", "RouteMapDict"]

459
src/fastmcp/server/plugins/openapi/provider.py Normal file
View file

@ -0,0 +1,459 @@
"""OpenAPIProvider for creating MCP components from OpenAPI specifications."""
from __future__ import annotations
from collections import Counter
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from typing import Any, Literal, cast
import httpx
from jsonschema_path import SchemaPath
from fastmcp.prompts import Prompt
from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.server.plugins.openapi.components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
_extract_mime_type_from_route,
_slugify,
)
from fastmcp.server.plugins.openapi.routing import (
DEFAULT_ROUTE_MAPPINGS,
ComponentFn,
MCPType,
RouteMap,
RouteMapFn,
_determine_route_type,
)
from fastmcp.server.providers.base import Provider
from fastmcp.tools.base import Tool
from fastmcp.utilities.components import FastMCPComponent
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.openapi import (
HTTPRoute,
extract_output_schema_from_responses,
parse_openapi_to_http_routes,
)
from fastmcp.utilities.openapi.director import RequestDirector
from fastmcp.utilities.versions import VersionSpec, version_sort_key
__all__ = [
"OpenAPIProvider",
]
logger = get_logger(__name__)
DEFAULT_TIMEOUT: float = 30.0
def resolve_spec_base_url(openapi_spec: dict[str, Any]) -> str:
"""Resolve the first `servers[0].url` in an OpenAPI spec, substituting
any `servers[0].variables[name].default` values into `{name}`
placeholders.
Raised to module level so callers that build their own httpx client
(e.g. the `OpenAPI` plugin applying user-configured headers/timeout)
can still honor spec server templates without duplicating the
substitution logic.
"""
servers = openapi_spec.get("servers", [])
if not servers or not servers[0].get("url"):
raise ValueError(
"No server URL found in OpenAPI spec. Either add a 'servers' "
"entry to the spec or provide an httpx.AsyncClient explicitly."
)
base_url = servers[0]["url"]
variables = servers[0].get("variables", {})
for name, var in variables.items():
base_url = base_url.replace(f"{{{name}}}", var.get("default", ""))
return base_url
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.plugins.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 | None = None,
*,
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,
validate_output: bool = True,
_owns_client: bool | None = None,
):
"""Initialize provider by parsing OpenAPI spec and creating components.
Args:
openapi_spec: OpenAPI schema as a dictionary
client: Optional httpx AsyncClient for making HTTP requests.
If not provided, a default client is created using the first
server URL from the OpenAPI spec with a 30-second timeout.
To customize timeout or other settings, pass your own client.
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
validate_output: If True (default), tools use the output schema
extracted from the OpenAPI spec for response validation. If
False, a permissive schema is used instead, allowing any
response structure while still returning structured JSON.
_owns_client: Private opt-in for callers (like the OpenAPI plugin)
that built `client` themselves and want the provider's lifespan
to close it on shutdown. Leave `None` for the default
"own it iff we built it here" behavior.
"""
super().__init__()
if _owns_client is None:
_owns_client = client is None
self._owns_client = _owns_client
if client is None:
client = self._create_default_client(openapi_spec)
self._client = client
self._mcp_component_fn = mcp_component_fn
self._validate_output = validate_output
# 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(cast(Any, openapi_spec))
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")
@classmethod
def _create_default_client(cls, openapi_spec: dict[str, Any]) -> httpx.AsyncClient:
"""Create a default httpx client from the OpenAPI spec's server URL."""
return httpx.AsyncClient(
base_url=resolve_spec_base_url(openapi_spec),
timeout=DEFAULT_TIMEOUT,
)
@asynccontextmanager
async def lifespan(self) -> AsyncIterator[None]:
"""Manage the lifecycle of the auto-created httpx client."""
if self._owns_client:
async with self._client:
yield
else:
yield
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,
)
if not self._validate_output and output_schema is not None:
# Use a permissive schema that accepts any object, preserving
# the wrap-result flag so non-object responses still get wrapped
permissive: dict[str, Any] = {
"type": "object",
"additionalProperties": True,
}
if output_schema.get("x-fastmcp-wrap-result"):
permissive["x-fastmcp-wrap-result"] = True
output_schema = permissive
tool_name = self._get_unique_name(name, "tool")
base_description = (
route.description
or route.summary
or f"Executes {route.method} {route.path}"
)
tool = OpenAPITool(
client=self._client,
route=route,
director=self._director,
name=tool_name,
description=base_description,
parameters=combined_schema,
output_schema=output_schema,
tags=set(route.tags or []) | tags,
)
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}"
)
resource = OpenAPIResource(
client=self._client,
route=route,
director=self._director,
uri=resource_uri,
name=resource_name,
description=base_description,
mime_type=_extract_mime_type_from_route(route),
tags=set(route.tags or []) | tags,
)
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}"
)
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=base_description,
parameters=template_params_schema,
tags=set(route.tags or []) | tags,
mime_type=_extract_mime_type_from_route(route),
)
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, version: VersionSpec | None = None
) -> Tool | None:
"""Get a tool by name."""
tool = self._tools.get(name)
if tool is None:
return None
if version is not None and not version.matches(tool.version):
return None
return tool
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, version: VersionSpec | None = None
) -> Resource | None:
"""Get a resource by URI."""
resource = self._resources.get(uri)
if resource is None:
return None
if version is not None and not version.matches(resource.version):
return None
return resource
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, version: VersionSpec | None = None
) -> ResourceTemplate | None:
"""Get a resource template that matches the given URI."""
matching = [t for t in self._templates.values() if t.matches(uri) is not None]
if not matching:
return None
if version is not None:
matching = [t for t in matching if version.matches(t.version)]
if not matching:
return None
return max(matching, key=version_sort_key) # type: ignore[type-var] # ty:ignore[invalid-return-type]
async def _list_prompts(self) -> Sequence[Prompt]:
"""Return empty list - OpenAPI doesn't create prompts."""
return []
async def get_tasks(self) -> Sequence[FastMCPComponent]:
"""Return empty list - OpenAPI components don't support tasks."""
return []

109
src/fastmcp/server/plugins/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.plugins.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)

32
src/fastmcp/server/providers/openapi/__init__.py
View file

@ -1,26 +1,32 @@
"""OpenAPI provider for FastMCP.
"""Backwards-compatibility shim — OpenAPI moved to `fastmcp.server.plugins.openapi`.
This module provides OpenAPI integration for FastMCP through the Provider pattern.
The preferred entry point is now the `OpenAPI` plugin:
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
import httpx
from fastmcp.server.plugins.openapi import OpenAPI, OpenAPIConfig
client = httpx.AsyncClient(base_url="https://api.example.com")
provider = OpenAPIProvider(openapi_spec=spec, client=client)
mcp = FastMCP("API Server", providers=[provider])
```
mcp = FastMCP("Server", plugins=[OpenAPI(OpenAPIConfig(spec=...))])
`OpenAPIProvider` and its helpers (`RouteMap`, `MCPType`, component
classes) remain importable from this package for direct composition.
Importing from this top-level path does **not** emit a deprecation
warning — it stays silent so that unrelated code in fastmcp that
happens to touch `fastmcp.server.providers.openapi` doesn't spray
warnings. Users who import from the leaf submodules (`.provider`,
`.routing`, `.components`) directly will see a `FastMCPDeprecationWarning`
pointing at the new location.
"""
from fastmcp.server.providers.openapi.components import (
# Silent passthrough at the package level — re-export from the new
# location directly so neither this import nor the lazy provider import
# inside `fastmcp.server.providers.__init__` fires a deprecation warning.
from fastmcp.server.plugins.openapi.components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
)
from fastmcp.server.providers.openapi.provider import OpenAPIProvider
from fastmcp.server.providers.openapi.routing import (
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
from fastmcp.server.plugins.openapi.routing import (
ComponentFn,
MCPType,
RouteMap,

424
src/fastmcp/server/providers/openapi/components.py
View file

@ -1,53 +1,25 @@
"""OpenAPI component classes: Tool, Resource, and ResourceTemplate."""
"""Deprecation shim — OpenAPI component classes moved to
`fastmcp.server.plugins.openapi.components`.
"""
from __future__ import annotations
import json
import re
import warnings
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
import httpx
from mcp.types import ToolAnnotations
from pydantic.networks import AnyUrl
import fastmcp
from fastmcp.exceptions import FastMCPDeprecationWarning
from fastmcp.resources import (
Resource,
ResourceContent,
ResourceResult,
ResourceTemplate,
)
from fastmcp.server.dependencies import get_http_headers
from fastmcp.server.tasks.config import TaskConfig
from fastmcp.tools.base 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
_SAFE_HEADERS = frozenset(
{
"accept",
"accept-encoding",
"accept-language",
"cache-control",
"connection",
"content-length",
"content-type",
"host",
"user-agent",
}
from fastmcp.server.plugins.openapi.components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
_extract_mime_type_from_route,
)
def _redact_headers(headers: httpx.Headers) -> dict[str, str]:
return {k: v if k.lower() in _SAFE_HEADERS else "***" for k, v in headers.items()}
warnings.warn(
"fastmcp.server.providers.openapi.components has moved to "
"fastmcp.server.plugins.openapi.components. Prefer the OpenAPI "
"plugin: `from fastmcp.server.plugins.openapi import OpenAPI`. This "
"old leaf-submodule import path will be removed in a future release.",
FastMCPDeprecationWarning,
stacklevel=2,
)
__all__ = [
"OpenAPIResource",
@ -55,367 +27,3 @@ __all__ = [
"OpenAPITool",
"_extract_mime_type_from_route",
]
logger = get_logger(__name__)
# Default MIME type when no response content type can be inferred
_DEFAULT_MIME_TYPE = "application/json"
def _extract_mime_type_from_route(route: HTTPRoute) -> str:
"""Extract the primary MIME type from an HTTPRoute's response definitions.
Looks for the first successful response (2xx) and returns its content type.
Prefers JSON-compatible types when multiple are available.
Falls back to "application/json" when no response content type is declared.
"""
if not route.responses:
return _DEFAULT_MIME_TYPE
# Priority order for success status codes
success_codes = ["200", "201", "202", "204"]
response_info = None
for status_code in success_codes:
if status_code in route.responses:
response_info = route.responses[status_code]
break
# If no explicit success codes, try any 2xx response
if response_info is None:
for status_code, resp_info in route.responses.items():
if status_code.startswith("2"):
response_info = resp_info
break
if response_info is None or not response_info.content_schema:
return _DEFAULT_MIME_TYPE
# If there's only one content type, use it directly
content_types = list(response_info.content_schema.keys())
if len(content_types) == 1:
return content_types[0]
# When multiple types exist, prefer JSON-compatible types
json_compatible_types = [
"application/json",
"application/vnd.api+json",
"application/hal+json",
"application/ld+json",
"text/json",
]
for ct in json_compatible_types:
if ct in response_info.content_schema:
return ct
# Fall back to the first available content type
return content_types[0]
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,
annotations: ToolAnnotations | None = None,
serializer: Callable[[Any], str] | None = None, # Deprecated
):
if serializer is not None and fastmcp.settings.deprecation_warnings:
warnings.warn(
"The `serializer` parameter is deprecated. "
"Return ToolResult from your tools for full control over serialization. "
"See https://gofastmcp.com/servers/tools#custom-serialization for migration examples.",
FastMCPDeprecationWarning,
stacklevel=2,
)
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
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."""
# Build the request — errors here are programming/schema issues,
# not HTTP failures, so we catch them separately.
try:
base_url = str(self._client.base_url) or "http://localhost"
request = self._director.build(self._route, arguments, base_url)
if self._client.headers:
for key, value in self._client.headers.items():
if key not in request.headers:
request.headers[key] = value
mcp_headers = get_http_headers()
if mcp_headers:
for key, value in mcp_headers.items():
if key not in request.headers:
request.headers[key] = value
except Exception as e:
raise ValueError(
f"Error building request for {self._route.method.upper()} "
f"{self._route.path}: {type(e).__name__}: {e}"
) from e
# Send the request and process the response.
try:
logger.debug(
f"run - sending request; headers: {_redact_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
# Structured content must be a dict for the MCP protocol.
# Wrap non-dict values that slipped through (e.g. a backend
# returning an array when the schema declared an object).
if not isinstance(structured_output, dict):
structured_output = {"result": structured_output}
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.TimeoutException as e:
raise ValueError(f"HTTP request timed out ({type(e).__name__})") from e
except httpx.RequestError as e:
raise ValueError(f"Request error ({type(e).__name__}): {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,
):
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
def __repr__(self) -> str:
return f"OpenAPIResource(name={self.name!r}, uri={self.uri!r}, path={self._route.path})"
async def read(self) -> ResourceResult:
"""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,
)
response.raise_for_status()
content_type = response.headers.get("content-type", "").lower()
if "application/json" in content_type:
result = response.json()
return ResourceResult(
contents=[
ResourceContent(
content=json.dumps(result), mime_type="application/json"
)
]
)
elif any(ct in content_type for ct in ["text/", "application/xml"]):
return ResourceResult(
contents=[
ResourceContent(content=response.text, mime_type=self.mime_type)
]
)
else:
return ResourceResult(
contents=[
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.TimeoutException as e:
raise ValueError(f"HTTP request timed out ({type(e).__name__})") from e
except httpx.RequestError as e:
raise ValueError(f"Request error ({type(e).__name__}): {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,
mime_type: str = _DEFAULT_MIME_TYPE,
):
super().__init__(
uri_template=uri_template,
name=name,
description=description,
parameters=parameters,
tags=tags or set(),
mime_type=mime_type,
)
self._client = client
self._route = route
self._director = director
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=self.mime_type,
tags=set(self._route.tags or []),
)

447
src/fastmcp/server/providers/openapi/provider.py
View file

@ -1,436 +1,23 @@
"""OpenAPIProvider for creating MCP components from OpenAPI specifications."""
"""Deprecation shim — `OpenAPIProvider` moved to
`fastmcp.server.plugins.openapi.provider`.
from __future__ import annotations
Prefer the `OpenAPI` plugin at `fastmcp.server.plugins.openapi` for new
code. `OpenAPIProvider` is still importable here for backcompat with
callers that composed it directly.
"""
from collections import Counter
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from typing import Any, Literal, cast
import warnings
import httpx
from jsonschema_path import SchemaPath
from fastmcp.exceptions import FastMCPDeprecationWarning
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
from fastmcp.prompts import Prompt
from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.server.providers.base import Provider
from fastmcp.server.providers.openapi.components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
_extract_mime_type_from_route,
_slugify,
warnings.warn(
"fastmcp.server.providers.openapi.provider has moved to "
"fastmcp.server.plugins.openapi.provider. Prefer the OpenAPI plugin: "
"`from fastmcp.server.plugins.openapi import OpenAPI`. This old "
"leaf-submodule import path will be removed in a future release.",
FastMCPDeprecationWarning,
stacklevel=2,
)
from fastmcp.server.providers.openapi.routing import (
DEFAULT_ROUTE_MAPPINGS,
ComponentFn,
MCPType,
RouteMap,
RouteMapFn,
_determine_route_type,
)
from fastmcp.tools.base import Tool
from fastmcp.utilities.components import FastMCPComponent
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.openapi import (
HTTPRoute,
extract_output_schema_from_responses,
parse_openapi_to_http_routes,
)
from fastmcp.utilities.openapi.director import RequestDirector
from fastmcp.utilities.versions import VersionSpec, version_sort_key
__all__ = [
"OpenAPIProvider",
]
logger = get_logger(__name__)
DEFAULT_TIMEOUT: float = 30.0
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 | None = None,
*,
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,
validate_output: bool = True,
):
"""Initialize provider by parsing OpenAPI spec and creating components.
Args:
openapi_spec: OpenAPI schema as a dictionary
client: Optional httpx AsyncClient for making HTTP requests.
If not provided, a default client is created using the first
server URL from the OpenAPI spec with a 30-second timeout.
To customize timeout or other settings, pass your own client.
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
validate_output: If True (default), tools use the output schema
extracted from the OpenAPI spec for response validation. If
False, a permissive schema is used instead, allowing any
response structure while still returning structured JSON.
"""
super().__init__()
self._owns_client = client is None
if client is None:
client = self._create_default_client(openapi_spec)
self._client = client
self._mcp_component_fn = mcp_component_fn
self._validate_output = validate_output
# 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(cast(Any, openapi_spec))
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")
@classmethod
def _create_default_client(cls, openapi_spec: dict[str, Any]) -> httpx.AsyncClient:
"""Create a default httpx client from the OpenAPI spec's server URL."""
servers = openapi_spec.get("servers", [])
if not servers or not servers[0].get("url"):
raise ValueError(
"No server URL found in OpenAPI spec. Either add a 'servers' "
"entry to the spec or provide an httpx.AsyncClient explicitly."
)
base_url = servers[0]["url"]
variables = servers[0].get("variables", {})
for name, var in variables.items():
base_url = base_url.replace(f"{{{name}}}", var.get("default", ""))
return httpx.AsyncClient(base_url=base_url, timeout=DEFAULT_TIMEOUT)
@asynccontextmanager
async def lifespan(self) -> AsyncIterator[None]:
"""Manage the lifecycle of the auto-created httpx client."""
if self._owns_client:
async with self._client:
yield
else:
yield
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,
)
if not self._validate_output and output_schema is not None:
# Use a permissive schema that accepts any object, preserving
# the wrap-result flag so non-object responses still get wrapped
permissive: dict[str, Any] = {
"type": "object",
"additionalProperties": True,
}
if output_schema.get("x-fastmcp-wrap-result"):
permissive["x-fastmcp-wrap-result"] = True
output_schema = permissive
tool_name = self._get_unique_name(name, "tool")
base_description = (
route.description
or route.summary
or f"Executes {route.method} {route.path}"
)
tool = OpenAPITool(
client=self._client,
route=route,
director=self._director,
name=tool_name,
description=base_description,
parameters=combined_schema,
output_schema=output_schema,
tags=set(route.tags or []) | tags,
)
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}"
)
resource = OpenAPIResource(
client=self._client,
route=route,
director=self._director,
uri=resource_uri,
name=resource_name,
description=base_description,
mime_type=_extract_mime_type_from_route(route),
tags=set(route.tags or []) | tags,
)
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}"
)
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=base_description,
parameters=template_params_schema,
tags=set(route.tags or []) | tags,
mime_type=_extract_mime_type_from_route(route),
)
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, version: VersionSpec | None = None
) -> Tool | None:
"""Get a tool by name."""
tool = self._tools.get(name)
if tool is None:
return None
if version is not None and not version.matches(tool.version):
return None
return tool
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, version: VersionSpec | None = None
) -> Resource | None:
"""Get a resource by URI."""
resource = self._resources.get(uri)
if resource is None:
return None
if version is not None and not version.matches(resource.version):
return None
return resource
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, version: VersionSpec | None = None
) -> ResourceTemplate | None:
"""Get a resource template that matches the given URI."""
matching = [t for t in self._templates.values() if t.matches(uri) is not None]
if not matching:
return None
if version is not None:
matching = [t for t in matching if version.matches(t.version)]
if not matching:
return None
return max(matching, key=version_sort_key) # type: ignore[type-var] # ty:ignore[invalid-return-type]
async def _list_prompts(self) -> Sequence[Prompt]:
"""Return empty list - OpenAPI doesn't create prompts."""
return []
async def get_tasks(self) -> Sequence[FastMCPComponent]:
"""Return empty list - OpenAPI components don't support tasks."""
return []
__all__ = ["OpenAPIProvider"]

118
src/fastmcp/server/providers/openapi/routing.py
View file

@ -1,23 +1,25 @@
"""Route mapping logic for OpenAPI operations."""
"""Deprecation shim — OpenAPI route-mapping types moved to
`fastmcp.server.plugins.openapi.routing`.
"""
from __future__ import annotations
import warnings
import enum
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from re import Pattern
from typing import TYPE_CHECKING, Literal
from fastmcp.exceptions import FastMCPDeprecationWarning
from fastmcp.server.plugins.openapi.routing import (
ComponentFn,
MCPType,
RouteMap,
RouteMapFn,
)
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
warnings.warn(
"fastmcp.server.providers.openapi.routing has moved to "
"fastmcp.server.plugins.openapi.routing. Prefer the OpenAPI plugin: "
"`from fastmcp.server.plugins.openapi import OpenAPI`. This old "
"leaf-submodule import path will be removed in a future release.",
FastMCPDeprecationWarning,
stacklevel=2,
)
__all__ = [
"ComponentFn",
@ -25,85 +27,3 @@ __all__ = [
"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)

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

@ -98,9 +98,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.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.plugins.openapi import RouteMap
from fastmcp.server.plugins.openapi.routing import ComponentFn as OpenAPIComponentFn
from fastmcp.server.plugins.openapi.routing import RouteMapFn as OpenAPIRouteMapFn
from fastmcp.server.providers.proxy import FastMCPProxy
logger = get_logger(__name__)
@ -2495,21 +2495,28 @@ class FastMCP(
**settings: Additional settings passed to FastMCP
Returns:
A FastMCP server with an OpenAPIProvider attached.
A FastMCP server with the OpenAPI plugin attached.
"""
from .providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi import OpenAPI, OpenAPIConfig
provider: Provider = OpenAPIProvider(
openapi_spec=openapi_spec,
# `from_openapi` returns an eagerly-populated server (callers
# frequently inspect `list_tools()` before running the server).
# Build the plugin to reuse its config-validation and provider-
# construction logic, then extract the provider eagerly rather
# than deferring to plugin-lifespan contribution.
plugin = OpenAPI(
OpenAPIConfig(
spec=openapi_spec,
mcp_names=mcp_names,
tags=sorted(tags) if tags else [],
validate_output=validate_output,
),
client=client,
route_maps=route_maps,
route_map_fn=route_map_fn,
mcp_component_fn=mcp_component_fn,
mcp_names=mcp_names,
tags=tags,
validate_output=validate_output,
)
return cls(name=name, providers=[provider], **settings)
return cls(name=name, providers=list(plugin.providers()), **settings)
@classmethod
def from_fastapi(
@ -2540,9 +2547,9 @@ class FastMCP(
**settings: Additional settings passed to FastMCP
Returns:
A FastMCP server with an OpenAPIProvider attached.
A FastMCP server with the OpenAPI plugin attached.
"""
from .providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi import OpenAPI, OpenAPIConfig
if httpx_client_kwargs is None:
httpx_client_kwargs = {}
@ -2555,16 +2562,18 @@ class FastMCP(
server_name = name or app.title
provider: Provider = OpenAPIProvider(
openapi_spec=app.openapi(),
plugin = OpenAPI(
OpenAPIConfig(
spec=app.openapi(),
mcp_names=mcp_names,
tags=sorted(tags) if tags else [],
),
client=client,
route_maps=route_maps,
route_map_fn=route_map_fn,
mcp_component_fn=mcp_component_fn,
mcp_names=mcp_names,
tags=tags,
)
return cls(name=server_name, providers=[provider], **settings)
return cls(name=server_name, providers=list(plugin.providers()), **settings)
@classmethod
def as_proxy(

8
tests/deprecated/openapi/test_openapi.py
View file

@ -27,7 +27,7 @@ class TestDeprecatedServerOpenAPIImports:
x for x in w if issubclass(x.category, DeprecationWarning)
]
assert len(deprecation_warnings) >= 1
assert "providers.openapi" in str(deprecation_warnings[0].message)
assert "plugins.openapi" in str(deprecation_warnings[0].message)
def test_import_routing_emits_warning(self):
"""Importing from fastmcp.server.openapi.routing should emit deprecation warning."""
@ -43,7 +43,7 @@ class TestDeprecatedServerOpenAPIImports:
x for x in w if issubclass(x.category, DeprecationWarning)
]
assert len(deprecation_warnings) >= 1
assert "providers.openapi" in str(deprecation_warnings[0].message)
assert "plugins.openapi" in str(deprecation_warnings[0].message)
def test_fastmcp_openapi_class_emits_warning(self):
"""Using FastMCPOpenAPI should emit deprecation warning."""
@ -117,7 +117,7 @@ class TestDeprecatedExperimentalOpenAPIImports:
x for x in w if issubclass(x.category, DeprecationWarning)
]
assert len(deprecation_warnings) >= 1
assert "providers.openapi" in str(deprecation_warnings[0].message)
assert "plugins.openapi" in str(deprecation_warnings[0].message)
def test_experimental_imports_still_work(self):
"""All expected symbols should be importable from experimental."""
@ -152,7 +152,7 @@ class TestDeprecatedComponentsImports:
x for x in w if issubclass(x.category, DeprecationWarning)
]
assert len(deprecation_warnings) >= 1
assert "providers.openapi" in str(deprecation_warnings[0].message)
assert "plugins.openapi" in str(deprecation_warnings[0].message)
def test_components_imports_still_work(self):
"""Component classes should be importable from deprecated location."""

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

4
tests/server/providers/openapi/test_comprehensive.py → tests/server/plugins/openapi/test_comprehensive.py
View file

@ -9,7 +9,7 @@ from httpx import Response
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
def create_openapi_server(
@ -929,7 +929,7 @@ class TestOpenAPIPostEdgeCases:
async def test_unexpected_error_in_request_building_gives_useful_message(self):
"""Unexpected exceptions during request building should produce useful errors."""
from fastmcp.server.providers.openapi.components import OpenAPITool
from fastmcp.server.plugins.openapi.components import OpenAPITool
from fastmcp.utilities.openapi.director import RequestDirector
from fastmcp.utilities.openapi.models import HTTPRoute

2
tests/server/providers/openapi/test_deepobject_style.py → tests/server/plugins/openapi/test_deepobject_style.py
View file

@ -5,7 +5,7 @@ import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
def create_openapi_server(

2
tests/server/providers/openapi/test_end_to_end_compatibility.py → tests/server/plugins/openapi/test_end_to_end_compatibility.py
View file

@ -5,7 +5,7 @@ import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
def create_openapi_server(

6
tests/server/providers/openapi/test_openapi_features.py → tests/server/plugins/openapi/test_openapi_features.py
View file

@ -8,12 +8,12 @@ from httpx import Response
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.providers.openapi.components import (
from fastmcp.server.plugins.openapi.components import (
_extract_mime_type_from_route,
_redact_headers,
)
from fastmcp.server.providers.openapi.routing import MCPType, RouteMap
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
from fastmcp.server.plugins.openapi.routing import MCPType, RouteMap
from fastmcp.utilities.openapi.models import HTTPRoute, ResponseInfo

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

2
tests/server/providers/openapi/test_parameter_collisions.py → tests/server/plugins/openapi/test_parameter_collisions.py
View file

@ -5,7 +5,7 @@ import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
def create_openapi_server(

2
tests/server/providers/openapi/test_performance_comparison.py → tests/server/plugins/openapi/test_performance_comparison.py
View file

@ -7,7 +7,7 @@ import httpx
import pytest
from fastmcp import FastMCP
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
def create_openapi_server(

3
tests/server/providers/openapi/test_server.py → tests/server/plugins/openapi/test_server.py
View file

@ -5,8 +5,7 @@ import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.providers.openapi import OpenAPIProvider
from fastmcp.server.providers.openapi.provider import DEFAULT_TIMEOUT
from fastmcp.server.plugins.openapi.provider import DEFAULT_TIMEOUT, OpenAPIProvider
class TestOpenAPIProviderServerVariables:

286
tests/server/plugins/test_openapi_plugin.py Normal file
View file

@ -0,0 +1,286 @@
"""Tests for the OpenAPI plugin wrapper.
Transform/provider behavior is covered by the existing OpenAPIProvider
tests in `tests/server/providers/openapi/`. This file only covers
plugin-layer concerns — config validation, dict→RouteMap conversion,
spec_path loading, and the escape-hatch wiring.
"""
from __future__ import annotations
import json
from pathlib import Path
import httpx
import pytest
from pydantic import ValidationError
from fastmcp import Client, FastMCP
from fastmcp.server.plugins.openapi import MCPType, OpenAPI, OpenAPIConfig, RouteMap
from fastmcp.server.plugins.openapi.plugin import RouteMapDict
from fastmcp.server.plugins.openapi.provider import OpenAPIProvider
PETSTORE_SPEC: dict = {
"openapi": "3.0.0",
"info": {"title": "Petstore", "version": "1.0"},
"servers": [{"url": "https://petstore.example.com"}],
"paths": {
"/pets": {
"get": {
"operationId": "list_pets",
"responses": {"200": {"description": "ok"}},
},
"post": {
"operationId": "create_pet",
"responses": {"201": {"description": "created"}},
},
},
"/pets/{id}": {
"get": {
"operationId": "get_pet",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"responses": {"200": {"description": "ok"}},
}
},
},
}
class TestOpenAPIConfig:
def test_config_generic_binding(self):
assert OpenAPI._config_cls is OpenAPIConfig
def test_default_config_instantiable(self):
"""Defaults must pass the plugin framework's instantiate-with-no-args
contract. The spec/spec_path check fires at providers() time, not
at Config construction."""
assert OpenAPIConfig() # must not raise
def test_unknown_config_key_rejected(self):
with pytest.raises((ValidationError, Exception), match="forbid|extra"):
OpenAPIConfig(not_a_real_option=True) # ty: ignore[unknown-argument]
def test_meta_name_is_single_word(self):
"""'openapi' is one technical term — explicit meta override
prevents the kebab auto-deriver from producing 'open-api'."""
assert OpenAPI.meta.name == "openapi"
assert OpenAPI.meta.version is None
class TestSpecLoading:
async def test_inline_spec_builds_provider(self):
plugin = OpenAPI(OpenAPIConfig(spec=PETSTORE_SPEC))
mcp = FastMCP("petstore", plugins=[plugin])
async with Client(mcp) as c:
tools = await c.list_tools()
names = {t.name for t in tools}
assert {"list_pets", "create_pet", "get_pet"}.issubset(names)
async def test_spec_path_loads_from_disk(self, tmp_path: Path):
spec_file = tmp_path / "petstore.json"
spec_file.write_text(json.dumps(PETSTORE_SPEC))
plugin = OpenAPI(OpenAPIConfig(spec_path=str(spec_file)))
mcp = FastMCP("petstore", plugins=[plugin])
async with Client(mcp) as c:
tools = await c.list_tools()
names = {t.name for t in tools}
assert {"list_pets", "create_pet", "get_pet"}.issubset(names)
async def test_spec_path_loads_utf8_regardless_of_locale(self, tmp_path: Path):
"""Spec files must load as UTF-8, not via the process locale.
Otherwise a spec with non-ASCII descriptions (German umlauts,
Japanese, fancy quotes, etc.) fails on non-UTF-8 systems like
Windows cp1252 — see PR #4015 review thread."""
spec_with_unicode = {
**PETSTORE_SPEC,
"info": {"title": "Pëtstöre — 宠物商店", "version": "1.0"},
}
spec_file = tmp_path / "petstore-unicode.json"
spec_file.write_text(
json.dumps(spec_with_unicode, ensure_ascii=False),
encoding="utf-8",
)
plugin = OpenAPI(OpenAPIConfig(spec_path=str(spec_file)))
providers = plugin.providers()
assert isinstance(providers[0], OpenAPIProvider)
def test_missing_spec_fails_at_build_time(self):
plugin = OpenAPI(OpenAPIConfig())
with pytest.raises(ValueError, match="spec.*spec_path"):
plugin.providers()
def test_both_spec_and_spec_path_rejected(self, tmp_path: Path):
spec_file = tmp_path / "spec.json"
spec_file.write_text(json.dumps(PETSTORE_SPEC))
plugin = OpenAPI(OpenAPIConfig(spec=PETSTORE_SPEC, spec_path=str(spec_file)))
with pytest.raises(ValueError, match="exactly one"):
plugin.providers()
class TestRouteMapping:
def test_route_maps_dict_form_converts_to_typed(self):
plugin = OpenAPI(
OpenAPIConfig(
spec=PETSTORE_SPEC,
route_maps=[
RouteMapDict(
mcp_type="RESOURCE", methods=["GET"], pattern=r"^/pets$"
),
],
)
)
providers = plugin.providers()
assert isinstance(providers[0], OpenAPIProvider)
# The GET /pets route should have become a resource, not a tool.
async def test_list_pets_maps_to_resource_via_config(self):
plugin = OpenAPI(
OpenAPIConfig(
spec=PETSTORE_SPEC,
route_maps=[
RouteMapDict(
mcp_type="RESOURCE", methods=["GET"], pattern=r"^/pets$"
),
],
)
)
mcp = FastMCP("petstore", plugins=[plugin])
async with Client(mcp) as c:
tools = {t.name for t in await c.list_tools()}
resources = {str(r.uri) for r in await c.list_resources()}
assert "list_pets" not in tools
assert any("list_pets" in uri or "/pets" in uri for uri in resources)
def test_typed_route_maps_override_dict_config(self):
"""When users pass typed `route_maps=` to `__init__`, that beats
the dict form in Config — advanced users shouldn't be shadowed
by an empty default."""
plugin = OpenAPI(
OpenAPIConfig(spec=PETSTORE_SPEC),
route_maps=[RouteMap(mcp_type=MCPType.EXCLUDE, pattern=r".*")],
)
providers = plugin.providers()
provider = providers[0]
# Every route was excluded → provider has no tools/resources.
assert isinstance(provider, OpenAPIProvider)
class TestDefaultClient:
async def test_plugin_built_client_is_closed_on_provider_lifespan_exit(self):
"""When the plugin builds its own httpx client (user didn't pass
`client=`), the provider's lifespan must still close it on
shutdown. A leaked client was bug noted on PR #4015."""
plugin = OpenAPI(OpenAPIConfig(spec=PETSTORE_SPEC))
provider = plugin.providers()[0]
assert isinstance(provider, OpenAPIProvider)
client = provider._client
assert not client.is_closed
async with provider.lifespan():
pass
assert client.is_closed
async def test_server_variable_defaults_are_substituted(self):
"""Spec servers with `{variable}` placeholders must be resolved
using `servers[0].variables[name].default` before going to the
httpx client — otherwise the literal template leaks into every
request URL."""
templated_spec = {
**PETSTORE_SPEC,
"servers": [
{
"url": "https://{region}.api.example.com",
"variables": {"region": {"default": "us-east"}},
}
],
}
plugin = OpenAPI(OpenAPIConfig(spec=templated_spec))
provider = plugin.providers()[0]
assert isinstance(provider, OpenAPIProvider)
assert str(provider._client.base_url) == "https://us-east.api.example.com"
class TestEscapeHatches:
async def test_custom_client_is_used(self):
"""Passing `client=` bypasses the auto-derived httpx client."""
client = httpx.AsyncClient(base_url="https://override.example.com")
plugin = OpenAPI(OpenAPIConfig(spec=PETSTORE_SPEC), client=client)
providers = plugin.providers()
assert isinstance(providers[0], OpenAPIProvider)
# Access the provider's client through the known private attr.
# This is an implementation check — acceptable in a test.
assert providers[0]._client is client
await client.aclose()
class TestDeprecationShim:
"""The old `fastmcp.server.providers.openapi` location now shims
back to the new plugin package. Top-level import is silent (so
unrelated code touching `fastmcp.server.providers` doesn't spray
warnings), but leaf submodules emit a `FastMCPDeprecationWarning`."""
async def test_top_level_old_path_is_silent_and_functional(self):
"""Still-common `from fastmcp.server.providers.openapi import
OpenAPIProvider` keeps working without emitting a warning."""
import warnings
from fastmcp.exceptions import FastMCPDeprecationWarning
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
from fastmcp.server.providers.openapi import (
OpenAPIProvider as LegacyProvider,
)
fastmcp_warns = [
w for w in caught if issubclass(w.category, FastMCPDeprecationWarning)
]
assert not fastmcp_warns
client = httpx.AsyncClient(base_url="https://petstore.example.com")
provider = LegacyProvider(openapi_spec=PETSTORE_SPEC, client=client)
mcp = FastMCP("petstore", providers=[provider])
async with Client(mcp) as c:
tools = {t.name for t in await c.list_tools()}
assert {"list_pets", "create_pet", "get_pet"}.issubset(tools)
assert LegacyProvider is OpenAPIProvider
await client.aclose()
def test_leaf_submodule_import_emits_deprecation_warning(self):
import importlib
import sys
import warnings
from fastmcp.exceptions import FastMCPDeprecationWarning
sys.modules.pop("fastmcp.server.providers.openapi.provider", None)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
importlib.import_module("fastmcp.server.providers.openapi.provider")
fastmcp_warns = [
w for w in caught if issubclass(w.category, FastMCPDeprecationWarning)
]
assert any("plugins.openapi" in str(w.message) for w in fastmcp_warns), (
f"expected FastMCPDeprecationWarning pointing at plugins.openapi, "
f"got {[(w.category.__name__, str(w.message)) for w in caught]}"
)