Spaces:
Running
Running
File size: 16,099 Bytes
8c5bdf2 4394198 8c5bdf2 3ce4c1b 8c5bdf2 4394198 8c5bdf2 eafc773 8c5bdf2 eafc773 8c5bdf2 3ce4c1b 8c5bdf2 3ce4c1b 8c5bdf2 3ce4c1b 8c5bdf2 3ce4c1b 8c5bdf2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 | """FastMCP server implementation for OpenAPI integration."""
import re
from collections import Counter
from typing import Any, Literal
import httpx
from jsonschema_path import SchemaPath
# Import from our new utilities and components
from fastmcp.experimental.utilities.openapi import (
HTTPRoute,
extract_output_schema_from_responses,
format_simple_description,
parse_openapi_to_http_routes,
)
from fastmcp.experimental.utilities.openapi.director import RequestDirector
from fastmcp.server.server import FastMCP
from fastmcp.utilities.logging import get_logger
from .components import (
OpenAPIResource,
OpenAPIResourceTemplate,
OpenAPITool,
)
from .routing import (
DEFAULT_ROUTE_MAPPINGS,
ComponentFn,
MCPType,
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
class FastMCPOpenAPI(FastMCP):
"""
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.
Example:
```python
from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap, MCPType
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,
)
```
"""
def __init__(
self,
openapi_spec: dict[str, Any],
client: httpx.AsyncClient,
name: str | 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,
timeout: float | None = None,
**settings: Any,
):
"""
Initialize a FastMCP server from an OpenAPI schema.
Args:
openapi_spec: OpenAPI schema as a dictionary or file path
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.
timeout: Optional timeout (in seconds) for all requests
**settings: Additional settings for FastMCP
"""
super().__init__(name=name or "OpenAPI FastMCP", **settings)
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[:-1]}. "
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,
)
# Get a unique tool name
tool_name = self._get_unique_name(name, "tool")
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
# Export public symbols
__all__ = [
"FastMCPOpenAPI",
]
|