diff --git a/docs/patterns/openapi.mdx b/docs/patterns/openapi.mdx index 166b5d89a..1767700c2 100644 --- a/docs/patterns/openapi.mdx +++ b/docs/patterns/openapi.mdx @@ -7,6 +7,8 @@ icon: code-branch If you have existing REST APIs documented with the OpenAPI Specification (OAS), FastMCP can automatically generate MCP tools, resources, and resource templates directly from that specification. This provides a quick way to make your existing HTTP APIs accessible to MCP clients and LLMs. +FastMCP supports both OpenAPI 3.0 and 3.1 specifications for maximum compatibility with existing API definitions. + ## The Goal: API -> MCP Server The core idea is to map OpenAPI paths and operations (like `GET /users/{id}` or `POST /orders`) to their corresponding MCP components: diff --git a/src/fastmcp/utilities/openapi.py b/src/fastmcp/utilities/openapi.py index 4427951b2..fcda41160 100644 --- a/src/fastmcp/utilities/openapi.py +++ b/src/fastmcp/utilities/openapi.py @@ -14,6 +14,16 @@ from openapi_pydantic import ( Response, Schema, ) + +# Import OpenAPI 3.0 models as well +from openapi_pydantic.v3.v3_0 import OpenAPI as OpenAPI_30 +from openapi_pydantic.v3.v3_0 import Operation as Operation_30 +from openapi_pydantic.v3.v3_0 import Parameter as Parameter_30 +from openapi_pydantic.v3.v3_0 import PathItem as PathItem_30 +from openapi_pydantic.v3.v3_0 import Reference as Reference_30 +from openapi_pydantic.v3.v3_0 import RequestBody as RequestBody_30 +from openapi_pydantic.v3.v3_0 import Response as Response_30 +from openapi_pydantic.v3.v3_0 import Schema as Schema_30 from pydantic import BaseModel, Field, ValidationError from fastmcp.utilities import openapi @@ -176,131 +186,6 @@ def _convert_to_parameter_location(param_in: str) -> ParameterLocation: return "query" -def _extract_parameters( - operation_params: list[Parameter | Reference] | None, - path_item_params: list[Parameter | Reference] | None, - openapi: OpenAPI, -) -> list[ParameterInfo]: - """Extracts and resolves parameters using corrected attribute names.""" - extracted_params: list[ParameterInfo] = [] - seen_params: dict[ - tuple[str, str], bool - ] = {} # Use string keys to avoid type issues - all_params_refs = (operation_params or []) + (path_item_params or []) - - for param_or_ref in all_params_refs: - try: - parameter = cast(Parameter, _resolve_ref(param_or_ref, openapi)) - if not isinstance(parameter, Parameter): - # ... (error logging remains the same) - continue - - # --- *** CORRECTED ATTRIBUTE ACCESS HERE *** --- - param_in = parameter.param_in # CORRECTED: Use 'param_in' - param_location = _convert_to_parameter_location(param_in) - param_schema_obj = parameter.param_schema # CORRECTED: Use 'param_schema' - # --- *** --- - - param_key = (parameter.name, param_in) - if param_key in seen_params: - continue - seen_params[param_key] = True - - param_schema_dict = {} - if param_schema_obj: # Check if schema exists - param_schema_dict = _extract_schema_as_dict(param_schema_obj, openapi) - elif parameter.content: - # Handle complex parameters with 'content' - first_media_type = next(iter(parameter.content.values()), None) - if ( - first_media_type and first_media_type.media_type_schema - ): # CORRECTED: Use 'media_type_schema' - param_schema_dict = _extract_schema_as_dict( - first_media_type.media_type_schema, openapi - ) - logger.debug( - f"Parameter '{parameter.name}' using schema from 'content' field." - ) - - # Manually create ParameterInfo instance using correct field names - param_info = ParameterInfo( - name=parameter.name, - location=param_location, # Use converted parameter location - required=parameter.required, - schema=param_schema_dict, # Populate 'schema' field in IR - description=parameter.description, - ) - extracted_params.append(param_info) - - except ( - ValidationError, - ValueError, - AttributeError, - TypeError, - ) as e: # Added TypeError - param_name = getattr( - param_or_ref, "name", getattr(param_or_ref, "ref", "unknown") - ) - logger.error( - f"Failed to extract parameter '{param_name}': {e}", exc_info=False - ) - - return extracted_params - - -def _extract_request_body( - request_body_or_ref: RequestBody | Reference | None, openapi: OpenAPI -) -> RequestBodyInfo | None: - """Extracts and resolves the request body using corrected attribute names.""" - if not request_body_or_ref: - return None - try: - request_body = cast(RequestBody, _resolve_ref(request_body_or_ref, openapi)) - if not isinstance(request_body, RequestBody): - # ... (error logging remains the same) - return None - - content_schemas: dict[str, JsonSchema] = {} - if request_body.content: - for media_type_str, media_type_obj in request_body.content.items(): - # --- *** CORRECTED ATTRIBUTE ACCESS HERE *** --- - if ( - isinstance(media_type_obj, MediaType) - and media_type_obj.media_type_schema - ): # CORRECTED: Use 'media_type_schema' - # --- *** --- - try: - # Use the corrected attribute here as well - schema_dict = _extract_schema_as_dict( - media_type_obj.media_type_schema, openapi - ) - content_schemas[media_type_str] = schema_dict - except ValueError as schema_err: - logger.error( - f"Failed to extract schema for media type '{media_type_str}' in request body: {schema_err}" - ) - elif not isinstance(media_type_obj, MediaType): - logger.warning( - f"Skipping invalid media type object for '{media_type_str}' (type: {type(media_type_obj)}) in request body." - ) - elif not media_type_obj.media_type_schema: # Corrected check - logger.warning( - f"Skipping media type '{media_type_str}' in request body because it lacks a schema." - ) - - return RequestBodyInfo( - required=request_body.required, - content_schema=content_schemas, - description=request_body.description, - ) - except (ValidationError, ValueError, AttributeError) as e: - ref_name = getattr(request_body_or_ref, "ref", "unknown") - logger.error( - f"Failed to extract request body '{ref_name}': {e}", exc_info=False - ) - return None - - def _extract_responses( operation_responses: dict[str, Response | Reference] | None, openapi: OpenAPI, @@ -358,194 +243,688 @@ def parse_openapi_to_http_routes(openapi_dict: dict[str, Any]) -> list[HTTPRoute """ Parses an OpenAPI schema dictionary into a list of HTTPRoute objects using the openapi-pydantic library. + + Supports both OpenAPI 3.0.x and 3.1.x versions. """ - routes: list[HTTPRoute] = [] + # Check OpenAPI version to use appropriate model + openapi_version = openapi_dict.get("openapi", "") + try: - openapi: OpenAPI = OpenAPI.model_validate(openapi_dict) - logger.info(f"Successfully parsed OpenAPI schema version: {openapi.openapi}") + if openapi_version.startswith("3.0"): + # Use OpenAPI 3.0 models + openapi_30 = OpenAPI_30.model_validate(openapi_dict) + logger.info( + f"Successfully parsed OpenAPI 3.0 schema version: {openapi_30.openapi}" + ) + parser = OpenAPI30Parser(openapi_30) + return parser.parse() + else: + # Default to OpenAPI 3.1 models + openapi_31 = OpenAPI.model_validate(openapi_dict) + logger.info( + f"Successfully parsed OpenAPI 3.1 schema version: {openapi_31.openapi}" + ) + parser = OpenAPI31Parser(openapi_31) + return parser.parse() except ValidationError as e: logger.error(f"OpenAPI schema validation failed: {e}") error_details = e.errors() logger.error(f"Validation errors: {error_details}") raise ValueError(f"Invalid OpenAPI schema: {error_details}") from e - if not openapi.paths: - logger.warning("OpenAPI schema has no paths defined.") - return [] - for path_str, path_item_obj in openapi.paths.items(): - if not isinstance(path_item_obj, PathItem): +# Base parser class for shared functionality +class BaseOpenAPIParser: + """Base class for OpenAPI parsers with common functionality.""" + + def _convert_to_parameter_location(self, param_in: str) -> ParameterLocation: + """Convert string parameter location to our ParameterLocation type.""" + if param_in == "path": + return "path" + elif param_in == "query": + return "query" + elif param_in == "header": + return "header" + elif param_in == "cookie": + return "cookie" + else: logger.warning( - f"Skipping invalid path item object for path '{path_str}' (type: {type(path_item_obj)})" + f"Unknown parameter location: {param_in}, defaulting to 'query'" ) - continue + return "query" - path_level_params = path_item_obj.parameters - # Iterate through possible HTTP methods defined in the PathItem model fields - # Use model_fields from the class, not the instance - for method_lower in PathItem.model_fields.keys(): - if method_lower not in [ - "get", - "put", - "post", - "delete", - "options", - "head", - "patch", - "trace", - ]: +class OpenAPI31Parser(BaseOpenAPIParser): + """Parser for OpenAPI 3.1 schemas.""" + + def __init__(self, openapi: OpenAPI): + self.openapi = openapi + + def parse(self) -> list[HTTPRoute]: + """Parse an OpenAPI 3.1 schema into HTTP routes.""" + routes: list[HTTPRoute] = [] + + if not self.openapi.paths: + logger.warning("OpenAPI schema has no paths defined.") + return [] + + for path_str, path_item_obj in self.openapi.paths.items(): + if not isinstance(path_item_obj, PathItem): + logger.warning( + f"Skipping invalid path item object for path '{path_str}' (type: {type(path_item_obj)})" + ) continue - operation: Operation | None = getattr(path_item_obj, method_lower, None) + path_level_params = path_item_obj.parameters - if operation and isinstance(operation, Operation): - method_upper = cast(HttpMethod, method_lower.upper()) - logger.debug(f"Processing operation: {method_upper} {path_str}") - try: - parameters = _extract_parameters( - operation.parameters, path_level_params, openapi + # Iterate through possible HTTP methods defined in the PathItem model fields + # Use model_fields from the class, not the instance + for method_lower in PathItem.model_fields.keys(): + if method_lower not in [ + "get", + "put", + "post", + "delete", + "options", + "head", + "patch", + "trace", + ]: + continue + + operation: Operation | None = getattr(path_item_obj, method_lower, None) + + if operation and isinstance(operation, Operation): + method_upper = cast(HttpMethod, method_lower.upper()) + logger.debug(f"Processing operation: {method_upper} {path_str}") + try: + parameters = self._extract_parameters( + operation.parameters, path_level_params + ) + request_body_info = self._extract_request_body( + operation.requestBody + ) + responses = self._extract_responses(operation.responses) + + route = HTTPRoute( + path=path_str, + method=method_upper, + operation_id=operation.operationId, + summary=operation.summary, + description=operation.description, + tags=operation.tags or [], + parameters=parameters, + request_body=request_body_info, + responses=responses, + ) + routes.append(route) + logger.info( + f"Successfully extracted route: {method_upper} {path_str}" + ) + except Exception as op_error: + op_id = operation.operationId or "unknown" + logger.error( + f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}", + exc_info=True, + ) + + logger.info(f"Finished parsing. Extracted {len(routes)} HTTP routes.") + return routes + + def _resolve_ref( + self, item: Reference | Schema | Parameter | RequestBody | Any + ) -> Any: + """Resolves a potential Reference object to its target definition.""" + if isinstance(item, Reference): + ref_str = item.ref + try: + if not ref_str.startswith("#/"): + raise ValueError( + f"External or non-local reference not supported: {ref_str}" ) - request_body_info = _extract_request_body( - operation.requestBody, openapi + parts = ref_str.strip("#/").split("/") + target = self.openapi + for part in parts: + if part.isdigit() and isinstance(target, list): + target = target[int(part)] + elif isinstance(target, BaseModel): + # Use model_extra for fields not explicitly defined (like components types) + # Check class fields first, then model_extra + if part in target.__class__.model_fields: + target = getattr(target, part, None) + elif target.model_extra and part in target.model_extra: + target = target.model_extra[part] + else: + # Special handling for components sub-types common structure + if part == "components" and hasattr(target, "components"): + target = getattr(target, "components") + elif hasattr(target, part): # Fallback check + target = getattr(target, part, None) + else: + target = None # Part not found + elif isinstance(target, dict): + target = target.get(part) + else: + raise ValueError( + f"Cannot traverse part '{part}' in reference '{ref_str}' from type {type(target)}" + ) + if target is None: + raise ValueError( + f"Reference part '{part}' not found in path '{ref_str}'" + ) + if isinstance(target, Reference): + return self._resolve_ref(target) + return target + except (AttributeError, KeyError, IndexError, TypeError, ValueError) as e: + raise ValueError(f"Failed to resolve reference '{ref_str}': {e}") from e + return item + + def _extract_schema_as_dict(self, schema_obj: Schema | Reference) -> JsonSchema: + """Resolves a schema/reference and returns it as a dictionary.""" + resolved_schema = self._resolve_ref(schema_obj) + if isinstance(resolved_schema, Schema): + # Using exclude_none=True might be better than exclude_unset sometimes + return resolved_schema.model_dump( + mode="json", by_alias=True, exclude_none=True + ) + elif isinstance(resolved_schema, dict): + logger.warning( + "Resolved schema reference resulted in a dict, not a Schema model." + ) + return resolved_schema + else: + ref_str = getattr(schema_obj, "ref", "unknown") + logger.warning( + f"Expected Schema after resolving ref '{ref_str}', got {type(resolved_schema)}. Returning empty dict." + ) + return {} + + def _extract_parameters( + self, + operation_params: list[Parameter | Reference] | None, + path_item_params: list[Parameter | Reference] | None, + ) -> list[ParameterInfo]: + """Extracts and resolves parameters using corrected attribute names.""" + extracted_params: list[ParameterInfo] = [] + seen_params: dict[ + tuple[str, str], bool + ] = {} # Use string keys to avoid type issues + all_params_refs = (operation_params or []) + (path_item_params or []) + + for param_or_ref in all_params_refs: + try: + parameter = cast(Parameter, self._resolve_ref(param_or_ref)) + if not isinstance(parameter, Parameter): + # ... (error logging remains the same) + continue + + # --- *** CORRECTED ATTRIBUTE ACCESS HERE *** --- + param_in = parameter.param_in # CORRECTED: Use 'param_in' + param_location = self._convert_to_parameter_location(param_in) + param_schema_obj = ( + parameter.param_schema + ) # CORRECTED: Use 'param_schema' + # --- *** --- + + param_key = (parameter.name, param_in) + if param_key in seen_params: + continue + seen_params[param_key] = True + + param_schema_dict = {} + if param_schema_obj: # Check if schema exists + param_schema_dict = self._extract_schema_as_dict(param_schema_obj) + elif parameter.content: + # Handle complex parameters with 'content' + first_media_type = next(iter(parameter.content.values()), None) + if ( + first_media_type and first_media_type.media_type_schema + ): # CORRECTED: Use 'media_type_schema' + param_schema_dict = self._extract_schema_as_dict( + first_media_type.media_type_schema + ) + logger.debug( + f"Parameter '{parameter.name}' using schema from 'content' field." + ) + + # Manually create ParameterInfo instance using correct field names + param_info = ParameterInfo( + name=parameter.name, + location=param_location, # Use converted parameter location + required=parameter.required, + schema=param_schema_dict, # Populate 'schema' field in IR + description=parameter.description, + ) + extracted_params.append(param_info) + + except ( + ValidationError, + ValueError, + AttributeError, + TypeError, + ) as e: # Added TypeError + param_name = getattr( + param_or_ref, "name", getattr(param_or_ref, "ref", "unknown") + ) + logger.error( + f"Failed to extract parameter '{param_name}': {e}", exc_info=False + ) + + return extracted_params + + def _extract_request_body( + self, request_body_or_ref: RequestBody | Reference | None + ) -> RequestBodyInfo | None: + """Extracts and resolves the request body using corrected attribute names.""" + if not request_body_or_ref: + return None + try: + request_body = cast(RequestBody, self._resolve_ref(request_body_or_ref)) + if not isinstance(request_body, RequestBody): + # ... (error logging remains the same) + return None + + content_schemas: dict[str, JsonSchema] = {} + if request_body.content: + for media_type_str, media_type_obj in request_body.content.items(): + # --- *** CORRECTED ATTRIBUTE ACCESS HERE *** --- + if ( + isinstance(media_type_obj, MediaType) + and media_type_obj.media_type_schema + ): # CORRECTED: Use 'media_type_schema' + # --- *** --- + try: + # Use the corrected attribute here as well + schema_dict = self._extract_schema_as_dict( + media_type_obj.media_type_schema + ) + content_schemas[media_type_str] = schema_dict + except ValueError as schema_err: + logger.error( + f"Failed to extract schema for media type '{media_type_str}' in request body: {schema_err}" + ) + elif not isinstance(media_type_obj, MediaType): + logger.warning( + f"Skipping invalid media type object for '{media_type_str}' (type: {type(media_type_obj)}) in request body." + ) + elif not media_type_obj.media_type_schema: # Corrected check + logger.warning( + f"Skipping media type '{media_type_str}' in request body because it lacks a schema." + ) + + return RequestBodyInfo( + required=request_body.required, + content_schema=content_schemas, + description=request_body.description, + ) + except (ValidationError, ValueError, AttributeError) as e: + ref_name = getattr(request_body_or_ref, "ref", "unknown") + logger.error( + f"Failed to extract request body '{ref_name}': {e}", exc_info=False + ) + return None + + def _extract_responses( + self, + operation_responses: dict[str, Response | Reference] | None, + ) -> dict[str, ResponseInfo]: + """Extracts and resolves response information for an operation.""" + extracted_responses: dict[str, ResponseInfo] = {} + if not operation_responses: + return extracted_responses + + for status_code, resp_or_ref in operation_responses.items(): + try: + response = cast(Response, self._resolve_ref(resp_or_ref)) + if not isinstance(response, Response): + ref_str = getattr(resp_or_ref, "ref", "unknown") + logger.warning( + f"Expected Response after resolving ref '{ref_str}' for status code {status_code}, got {type(response)}. Skipping." ) - responses = _extract_responses(operation.responses, openapi) + continue - route = HTTPRoute( - path=path_str, - method=method_upper, - operation_id=operation.operationId, - summary=operation.summary, - description=operation.description, - tags=operation.tags or [], - parameters=parameters, - request_body=request_body_info, - responses=responses, + content_schemas: dict[str, JsonSchema] = {} + if response.content: + for media_type_str, media_type_obj in response.content.items(): + if ( + isinstance(media_type_obj, MediaType) + and media_type_obj.media_type_schema + ): + try: + schema_dict = self._extract_schema_as_dict( + media_type_obj.media_type_schema + ) + content_schemas[media_type_str] = schema_dict + except ValueError as schema_err: + logger.error( + f"Failed to extract schema for media type '{media_type_str}' in response {status_code}: {schema_err}" + ) + + resp_info = ResponseInfo( + description=response.description, content_schema=content_schemas + ) + extracted_responses[str(status_code)] = resp_info + + except (ValidationError, ValueError, AttributeError) as e: + ref_name = getattr(resp_or_ref, "ref", "unknown") + logger.error( + f"Failed to extract response for status code {status_code} " + f"from reference '{ref_name}': {e}", + exc_info=False, + ) + + return extracted_responses + + +class OpenAPI30Parser(BaseOpenAPIParser): + """Parser for OpenAPI 3.0 schemas.""" + + def __init__(self, openapi: OpenAPI_30): + self.openapi = openapi + + def parse(self) -> list[HTTPRoute]: + """Parse an OpenAPI 3.0 schema into HTTP routes.""" + routes: list[HTTPRoute] = [] + + if not self.openapi.paths: + logger.warning("OpenAPI schema has no paths defined.") + return [] + + for path_str, path_item_obj in self.openapi.paths.items(): + if not isinstance(path_item_obj, PathItem_30): + logger.warning( + f"Skipping invalid path item object for path '{path_str}' (type: {type(path_item_obj)})" + ) + continue + + path_level_params = path_item_obj.parameters + + # Iterate through possible HTTP methods defined in the PathItem model fields + # Use model_fields from the class, not the instance + for method_lower in PathItem_30.model_fields.keys(): + if method_lower not in [ + "get", + "put", + "post", + "delete", + "options", + "head", + "patch", + "trace", + ]: + continue + + operation: Operation_30 | None = getattr( + path_item_obj, method_lower, None + ) + + if operation and isinstance(operation, Operation_30): + method_upper = cast(HttpMethod, method_lower.upper()) + logger.debug(f"Processing operation: {method_upper} {path_str}") + try: + parameters = self._extract_parameters( + operation.parameters, path_level_params + ) + request_body_info = self._extract_request_body( + operation.requestBody + ) + responses = self._extract_responses(operation.responses) + + route = HTTPRoute( + path=path_str, + method=method_upper, + operation_id=operation.operationId, + summary=operation.summary, + description=operation.description, + tags=operation.tags or [], + parameters=parameters, + request_body=request_body_info, + responses=responses, + ) + routes.append(route) + logger.info( + f"Successfully extracted route: {method_upper} {path_str}" + ) + except Exception as op_error: + op_id = operation.operationId or "unknown" + logger.error( + f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}", + exc_info=True, + ) + + logger.info(f"Finished parsing. Extracted {len(routes)} HTTP routes.") + return routes + + def _resolve_ref( + self, item: Reference_30 | Schema_30 | Parameter_30 | RequestBody_30 | Any + ) -> Any: + """Resolves a potential Reference object to its target definition for OpenAPI 3.0.""" + if isinstance(item, Reference_30): + ref_str = item.ref + try: + if not ref_str.startswith("#/"): + raise ValueError( + f"External or non-local reference not supported: {ref_str}" ) - routes.append(route) - logger.info( - f"Successfully extracted route: {method_upper} {path_str}" + parts = ref_str.strip("#/").split("/") + target = self.openapi + for part in parts: + if part.isdigit() and isinstance(target, list): + target = target[int(part)] + elif isinstance(target, BaseModel): + # Use model_extra for fields not explicitly defined (like components types) + # Check class fields first, then model_extra + if part in target.__class__.model_fields: + target = getattr(target, part, None) + elif target.model_extra and part in target.model_extra: + target = target.model_extra[part] + else: + # Special handling for components sub-types common structure + if part == "components" and hasattr(target, "components"): + target = getattr(target, "components") + elif hasattr(target, part): # Fallback check + target = getattr(target, part, None) + else: + target = None # Part not found + elif isinstance(target, dict): + target = target.get(part) + else: + raise ValueError( + f"Cannot traverse part '{part}' in reference '{ref_str}' from type {type(target)}" + ) + if target is None: + raise ValueError( + f"Reference part '{part}' not found in path '{ref_str}'" + ) + if isinstance(target, Reference_30): + return self._resolve_ref(target) + return target + except (AttributeError, KeyError, IndexError, TypeError, ValueError) as e: + raise ValueError(f"Failed to resolve reference '{ref_str}': {e}") from e + return item + + def _extract_schema_as_dict( + self, schema_obj: Schema_30 | Reference_30 + ) -> JsonSchema: + """Resolves a schema/reference and returns it as a dictionary for OpenAPI 3.0.""" + resolved_schema = self._resolve_ref(schema_obj) + if isinstance(resolved_schema, Schema_30): + # Using exclude_none=True might be better than exclude_unset sometimes + return resolved_schema.model_dump( + mode="json", by_alias=True, exclude_none=True + ) + elif isinstance(resolved_schema, dict): + logger.warning( + "Resolved schema reference resulted in a dict, not a Schema model." + ) + return resolved_schema + else: + ref_str = getattr(schema_obj, "ref", "unknown") + logger.warning( + f"Expected Schema after resolving ref '{ref_str}', got {type(resolved_schema)}. Returning empty dict." + ) + return {} + + def _extract_parameters( + self, + operation_params: list[Parameter_30 | Reference_30] | None, + path_item_params: list[Parameter_30 | Reference_30] | None, + ) -> list[ParameterInfo]: + """Extracts and resolves parameters for OpenAPI 3.0.""" + extracted_params: list[ParameterInfo] = [] + seen_params: dict[ + tuple[str, str], bool + ] = {} # Use string keys to avoid type issues + all_params_refs = (operation_params or []) + (path_item_params or []) + + for param_or_ref in all_params_refs: + try: + parameter = cast(Parameter_30, self._resolve_ref(param_or_ref)) + if not isinstance(parameter, Parameter_30): + logger.warning( + f"Expected Parameter after resolving reference, got {type(parameter)}. Skipping." ) - except Exception as op_error: - op_id = operation.operationId or "unknown" - logger.error( - f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}", - exc_info=True, + continue + + # OpenAPI 3.0 uses 'in' field for parameter location + param_in = parameter.param_in + param_location = self._convert_to_parameter_location(param_in) + param_schema_obj = parameter.param_schema + + param_key = (parameter.name, param_in) + if param_key in seen_params: + continue + seen_params[param_key] = True + + param_schema_dict = {} + if param_schema_obj: # Check if schema exists + param_schema_dict = self._extract_schema_as_dict(param_schema_obj) + elif parameter.content: + # Handle complex parameters with 'content' + first_media_type = next(iter(parameter.content.values()), None) + if first_media_type and first_media_type.media_type_schema: + param_schema_dict = self._extract_schema_as_dict( + first_media_type.media_type_schema + ) + logger.debug( + f"Parameter '{parameter.name}' using schema from 'content' field." + ) + + # Manually create ParameterInfo instance using correct field names + param_info = ParameterInfo( + name=parameter.name, + location=param_location, # Use converted parameter location + required=parameter.required, + schema=param_schema_dict, # Populate 'schema' field in IR + description=parameter.description, + ) + extracted_params.append(param_info) + + except ( + ValidationError, + ValueError, + AttributeError, + TypeError, + ) as e: # Added TypeError + param_name = getattr( + param_or_ref, "name", getattr(param_or_ref, "ref", "unknown") + ) + logger.error( + f"Failed to extract parameter '{param_name}': {e}", exc_info=False + ) + + return extracted_params + + def _extract_request_body( + self, request_body_or_ref: RequestBody_30 | Reference_30 | None + ) -> RequestBodyInfo | None: + """Extracts request body information for OpenAPI 3.0 using correct attribute names.""" + if request_body_or_ref is None: + return None + + try: + request_body = cast(RequestBody_30, self._resolve_ref(request_body_or_ref)) + + if not isinstance(request_body, RequestBody_30): + logger.warning( + f"Expected RequestBody after resolving reference, got {type(request_body)}. Returning None." + ) + return None + + request_body_info = RequestBodyInfo( + required=request_body.required, + description=request_body.description, + ) + + # Process content field for request body schemas + if request_body.content: + for media_type_key, media_type_obj in request_body.content.items(): + if ( + media_type_obj and media_type_obj.media_type_schema + ): # CORRECTED: Use 'media_type_schema' + schema_dict = self._extract_schema_as_dict( + media_type_obj.media_type_schema + ) + request_body_info.content_schema[media_type_key] = schema_dict + + return request_body_info + + except (ValidationError, ValueError, AttributeError) as e: + ref_str = getattr(request_body_or_ref, "ref", "unknown") + logger.error( + f"Failed to extract request body info from reference '{ref_str}': {e}", + exc_info=False, + ) + return None + + def _extract_responses( + self, + operation_responses: dict[str, Response_30 | Reference_30] | None, + ) -> dict[str, ResponseInfo]: + """Extracts response information from an OpenAPI 3.0 operation's responses.""" + extracted_responses: dict[str, ResponseInfo] = {} + if not operation_responses: + return extracted_responses + + for status_code, response_or_ref in operation_responses.items(): + try: + # Skip 'default' response for simplicity if needed + # if status_code == "default": + # continue + + response = cast(Response_30, self._resolve_ref(response_or_ref)) + + if not isinstance(response, Response_30): + logger.warning( + f"Expected Response after resolving reference for status code {status_code}, " + f"got {type(response)}. Skipping." ) + continue - logger.info(f"Finished parsing. Extracted {len(routes)} HTTP routes.") - return routes + response_info = ResponseInfo(description=response.description) + # Extract content schemas if present + if response.content: + for media_type_key, media_type_obj in response.content.items(): + if ( + media_type_obj and media_type_obj.media_type_schema + ): # CORRECTED: Use 'media_type_schema' + schema_dict = self._extract_schema_as_dict( + media_type_obj.media_type_schema + ) + response_info.content_schema[media_type_key] = schema_dict -# --- Example Usage (Optional) --- -if __name__ == "__main__": - import json + extracted_responses[status_code] = response_info - logging.basicConfig( - level=logging.INFO, format="%(levelname)s:%(name)s:%(message)s" - ) # Set to INFO + except (ValidationError, ValueError, AttributeError) as e: + ref_str = getattr(response_or_ref, "ref", "unknown") + logger.error( + f"Failed to extract response info for status code {status_code} " + f"from reference '{ref_str}': {e}", + exc_info=False, + ) - petstore_schema = { - "openapi": "3.1.0", # Keep corrected version - "info": {"title": "Simple Pet Store API", "version": "1.0.0"}, - "paths": { - "/pets": { - "get": { - "summary": "list all pets", - "operationId": "listPets", - "tags": ["pets"], - "parameters": [ - { - "name": "limit", - "in": "query", - "description": "How many items to return", - "required": False, - "schema": {"type": "integer", "format": "int32"}, - } - ], - "responses": {"200": {"description": "A paged array of pets"}}, - }, - "post": { - "summary": "Create a pet", - "operationId": "createPet", - "tags": ["pets"], - "requestBody": {"$ref": "#/components/requestBodies/PetBody"}, - "responses": {"201": {"description": "Null response"}}, - }, - }, - "/pets/{petId}": { - "get": { - "summary": "Info for a specific pet", - "operationId": "showPetById", - "tags": ["pets"], - "parameters": [ - { - "name": "petId", - "in": "path", - "required": True, - "description": "The id of the pet", - "schema": {"type": "string"}, - }, - { - "name": "X-Request-ID", - "in": "header", - "required": False, - "schema": {"type": "string", "format": "uuid"}, - }, - ], - "responses": {"200": {"description": "Information about the pet"}}, - }, - "parameters": [ # Path level parameter example - { - "name": "traceId", - "in": "header", - "description": "Common trace ID", - "required": False, - "schema": {"type": "string"}, - } - ], - }, - }, - "components": { - "schemas": { - "Pet": { - "type": "object", - "required": ["id", "name"], - "properties": { - "id": {"type": "integer", "format": "int64"}, - "name": {"type": "string"}, - "tag": {"type": "string"}, - }, - } - }, - "requestBodies": { - "PetBody": { - "description": "Pet object", - "required": True, - "content": { - "application/json": { - "schema": {"$ref": "#/components/schemas/Pet"} - } - }, - } - }, - }, - } - - print("--- Parsing Pet Store Schema using openapi-pydantic (Corrected) ---") - try: - http_routes = parse_openapi_to_http_routes(petstore_schema) - print(f"\n--- Extracted {len(http_routes)} Routes ---") - for i, route in enumerate(http_routes): - print(f"\nRoute {i + 1}:") - # Use model_dump for clean JSON-like output, show aliases from IR model - print( - json.dumps(route.model_dump(by_alias=True, exclude_none=True), indent=2) - ) # exclude_none is often cleaner - except ValueError as e: - print(f"\nError parsing schema: {e}") - except Exception as e: - print(f"\nAn unexpected error occurred: {e}") + return extracted_responses def clean_schema_for_display(schema: JsonSchema | None) -> JsonSchema | None: diff --git a/tests/utilities/openapi/test_openapi.py b/tests/utilities/openapi/test_openapi.py index 83cec52b3..e229990d1 100644 --- a/tests/utilities/openapi/test_openapi.py +++ b/tests/utilities/openapi/test_openapi.py @@ -311,6 +311,172 @@ def fastapi_route_map(parsed_fastapi_routes): } +@pytest.fixture +def openapi_30_schema() -> dict[str, Any]: + """Fixture that returns a simple OpenAPI 3.0.0 schema.""" + return { + "openapi": "3.0.0", + "info": {"title": "Simple API (OpenAPI 3.0)", "version": "1.0.0"}, + "paths": { + "/items": { + "get": { + "summary": "List all items", + "operationId": "listItems", + "parameters": [ + { + "name": "limit", + "in": "query", + "description": "How many items to return", + "required": False, + "schema": {"type": "integer"}, + } + ], + "responses": {"200": {"description": "A list of items"}}, + } + } + }, + } + + +@pytest.fixture +def openapi_31_schema() -> dict[str, Any]: + """Fixture that returns a simple OpenAPI 3.1.0 schema.""" + return { + "openapi": "3.1.0", + "info": {"title": "Simple API (OpenAPI 3.1)", "version": "1.0.0"}, + "paths": { + "/items": { + "get": { + "summary": "List all items", + "operationId": "listItems", + "parameters": [ + { + "name": "limit", + "in": "query", + "description": "How many items to return", + "required": False, + "schema": {"type": "integer"}, + } + ], + "responses": {"200": {"description": "A list of items"}}, + } + } + }, + } + + +@pytest.fixture +def openapi_30_with_references() -> dict[str, Any]: + """OpenAPI 3.0 schema with references to test resolution.""" + return { + "openapi": "3.0.0", + "info": {"title": "API with References (3.0)", "version": "1.0.0"}, + "paths": { + "/products": { + "post": { + "summary": "Create product", + "operationId": "createProduct", + "requestBody": { + "content": { + "application/json": { + "schema": {"$ref": "#/components/schemas/Product"} + } + }, + "required": True, + }, + "responses": { + "201": { + "description": "Product created", + "content": { + "application/json": { + "schema": {"$ref": "#/components/schemas/Product"} + } + }, + } + }, + } + } + }, + "components": { + "schemas": { + "Product": { + "type": "object", + "required": ["name", "price"], + "properties": { + "id": {"type": "string", "format": "uuid"}, + "name": {"type": "string"}, + "price": {"type": "number"}, + "category": {"$ref": "#/components/schemas/Category"}, + }, + }, + "Category": { + "type": "object", + "properties": { + "id": {"type": "integer"}, + "name": {"type": "string"}, + }, + }, + } + }, + } + + +@pytest.fixture +def openapi_31_with_references() -> dict[str, Any]: + """OpenAPI 3.1 schema with references to test resolution.""" + return { + "openapi": "3.1.0", + "info": {"title": "API with References (3.1)", "version": "1.0.0"}, + "paths": { + "/products": { + "post": { + "summary": "Create product", + "operationId": "createProduct", + "requestBody": { + "content": { + "application/json": { + "schema": {"$ref": "#/components/schemas/Product"} + } + }, + "required": True, + }, + "responses": { + "201": { + "description": "Product created", + "content": { + "application/json": { + "schema": {"$ref": "#/components/schemas/Product"} + } + }, + } + }, + } + } + }, + "components": { + "schemas": { + "Product": { + "type": "object", + "required": ["name", "price"], + "properties": { + "id": {"type": "string", "format": "uuid"}, + "name": {"type": "string"}, + "price": {"type": "number"}, + "category": {"$ref": "#/components/schemas/Category"}, + }, + }, + "Category": { + "type": "object", + "properties": { + "id": {"type": "integer"}, + "name": {"type": "string"}, + }, + }, + } + }, + } + + # --- Tests for PetStore schema --- # @@ -764,3 +930,166 @@ def test_fastapi_post_query_parameter_names(fastapi_route_map): param_names = [p.name for p in query_params] assert "file_name" in param_names assert "content_type" in param_names + + +def test_openapi_30_compatibility(openapi_30_schema): + """Test that OpenAPI 3.0 schemas can be parsed correctly.""" + # This will raise an exception if the parser doesn't support 3.0.0 + routes = parse_openapi_to_http_routes(openapi_30_schema) + + # Verify the route was parsed correctly + assert len(routes) == 1 + route = routes[0] + assert route.method == "GET" + assert route.path == "/items" + assert route.operation_id == "listItems" + assert len(route.parameters) == 1 + assert route.parameters[0].name == "limit" + + +def test_openapi_31_compatibility(openapi_31_schema): + """Test that OpenAPI 3.1 schemas can be parsed correctly.""" + routes = parse_openapi_to_http_routes(openapi_31_schema) + + # Verify the route was parsed correctly + assert len(routes) == 1 + route = routes[0] + assert route.method == "GET" + assert route.path == "/items" + assert route.operation_id == "listItems" + assert len(route.parameters) == 1 + assert route.parameters[0].name == "limit" + + +def test_version_detection_logic(): + """Test that the version detection logic correctly identifies 3.0 vs 3.1 schemas.""" + # Test 3.0 variations + for version in ["3.0.0", "3.0.1", "3.0.3"]: + schema = { + "openapi": version, + "info": {"title": "Test", "version": "1.0.0"}, + "paths": {}, + } + try: + parse_openapi_to_http_routes(schema) + # Expect no error + except Exception as e: + pytest.fail(f"Failed to parse OpenAPI {version} schema: {e}") + + # Test 3.1 variations + for version in ["3.1.0", "3.1.1"]: + schema = { + "openapi": version, + "info": {"title": "Test", "version": "1.0.0"}, + "paths": {}, + } + try: + parse_openapi_to_http_routes(schema) + # Expect no error + except Exception as e: + pytest.fail(f"Failed to parse OpenAPI {version} schema: {e}") + + +def test_openapi_30_reference_resolution(openapi_30_with_references): + """Test that references are correctly resolved in OpenAPI 3.0 schemas.""" + routes = parse_openapi_to_http_routes(openapi_30_with_references) + + assert len(routes) == 1 + route = routes[0] + assert route.method == "POST" + assert route.path == "/products" + + # Check request body + assert route.request_body is not None + assert route.request_body.required is True + assert "application/json" in route.request_body.content_schema + + # Check schema structure + json_schema = route.request_body.content_schema["application/json"] + assert json_schema["type"] == "object" + assert "properties" in json_schema + assert set(json_schema["required"]) == {"name", "price"} + + # Check primary fields are properly resolved + props = json_schema["properties"] + assert "id" in props + assert "name" in props + assert "price" in props + assert "category" in props + + # The category might be a reference or resolved object + category = props["category"] + # Either it's directly resolved with properties + # or it still has a $ref field + assert "properties" in category or "$ref" in category + + +def test_openapi_31_reference_resolution(openapi_31_with_references): + """Test that references are correctly resolved in OpenAPI 3.1 schemas.""" + routes = parse_openapi_to_http_routes(openapi_31_with_references) + + assert len(routes) == 1 + route = routes[0] + assert route.method == "POST" + assert route.path == "/products" + + # Check request body + assert route.request_body is not None + assert route.request_body.required is True + assert "application/json" in route.request_body.content_schema + + # Check schema structure + json_schema = route.request_body.content_schema["application/json"] + assert json_schema["type"] == "object" + assert "properties" in json_schema + assert set(json_schema["required"]) == {"name", "price"} + + # Check primary fields are properly resolved + props = json_schema["properties"] + assert "id" in props + assert "name" in props + assert "price" in props + assert "category" in props + + # The category might be a reference or resolved object + category = props["category"] + # Either it's directly resolved with properties + # or it still has a $ref field + assert "properties" in category or "$ref" in category + + +def test_consistent_output_across_versions( + openapi_30_with_references, openapi_31_with_references +): + """Test that both parsers produce equivalent output for equivalent schemas.""" + routes_30 = parse_openapi_to_http_routes(openapi_30_with_references) + routes_31 = parse_openapi_to_http_routes(openapi_31_with_references) + + # Convert to dict for easier comparison + route_30_dict = routes_30[0].model_dump(exclude_none=True) + route_31_dict = routes_31[0].model_dump(exclude_none=True) + + # They should be identical except for version-specific differences + # Compare path + assert route_30_dict["path"] == route_31_dict["path"] + # Compare method + assert route_30_dict["method"] == route_31_dict["method"] + # Compare operation_id + assert route_30_dict["operation_id"] == route_31_dict["operation_id"] + # Compare parameters + assert len(route_30_dict["parameters"]) == len(route_31_dict["parameters"]) + # Compare request body + assert ( + route_30_dict["request_body"]["required"] + == route_31_dict["request_body"]["required"] + ) + # Compare response structure + assert "201" in route_30_dict["responses"] and "201" in route_31_dict["responses"] + # The schemas should contain the same essential fields + schema_30 = route_30_dict["request_body"]["content_schema"]["application/json"][ + "properties" + ] + schema_31 = route_31_dict["request_body"]["content_schema"]["application/json"][ + "properties" + ] + assert set(schema_30.keys()) == set(schema_31.keys())