from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any, Set
from swagger_mcp.endpoint import Endpoint
from swagger_mcp.logging import setup_logger
logger = setup_logger(__name__)
@dataclass
class SimpleEndpoint:
"""
A simplified representation of an API endpoint where path parameters,
query parameters, and request body are combined into a single parameter object.
"""
# Basic endpoint information
path: str
method: str
operation_id: str
summary: str
description: str = ""
deprecated: bool = False
# Server information
servers: List[Dict[str, Any]] = field(default_factory=list)
# Authentication and security
requires_bearer_auth: bool = False
requires_oauth: bool = False # Indicates if OAuth flow authentication is required
security_requirements: List[Dict[str, List[str]]] = field(default_factory=list)
oauth_scopes: List[str] = field(default_factory=list) # OAuth scopes required for this endpoint
# Combined parameter information
combined_parameter_schema: Dict[str, Any] = field(default_factory=dict)
# Parameter type mapping
parameter_type_mapping: Dict[str, str] = field(default_factory=dict)
# Content type information
request_content_types: List[str] = field(default_factory=list)
response_content_types: List[str] = field(default_factory=list)
# Response information
responses: Dict[str, Dict[str, Any]] = field(default_factory=dict)
# Additional metadata
tags: List[str] = field(default_factory=list)
@property
def endpoint_key(self) -> str:
"""
Generate a unique key for this endpoint based on method and path.
Returns:
A string in the format "METHOD /path"
"""
return f"{self.method} {self.path}"
@property
def default_server_url(self) -> Optional[str]:
"""
Get the default server URL for this endpoint, if available.
Returns:
The URL string or None if no servers are defined
"""
if self.servers and 'url' in self.servers[0]:
return self.servers[0]['url']
return None
def get_full_url(self, server_url: Optional[str] = None, params: Optional[Dict[str, Any]] = None) -> str:
"""
Construct the full URL for this endpoint with path parameters substituted.
Args:
server_url: Server URL to use (defaults to the first server in the servers list)
params: Dictionary of combined parameters (path parameters will be extracted)
Returns:
The complete URL string
"""
base_url = server_url or self.default_server_url or ""
# Remove trailing slash from base URL if present
if base_url and base_url.endswith('/'):
base_url = base_url[:-1]
# Ensure path starts with slash
endpoint_path = self.path
if not endpoint_path.startswith('/'):
endpoint_path = '/' + endpoint_path
# Substitute path parameters if provided
if params:
path_params = {k: v for k, v in params.items()
if k in self.parameter_type_mapping and self.parameter_type_mapping[k] == 'path'}
for param_name, param_value in path_params.items():
placeholder = '{' + param_name + '}'
endpoint_path = endpoint_path.replace(placeholder, str(param_value))
return f"{base_url}{endpoint_path}"
def get_required_parameters(self) -> Set[str]:
"""
Get a set of all required parameters.
Returns:
Set of required parameter names
"""
required_params = set()
if 'required' in self.combined_parameter_schema:
required_params = set(self.combined_parameter_schema['required'])
return required_params
def get_path_parameters(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract path parameters from the combined parameters.
Args:
params: Combined parameters dictionary
Returns:
Dictionary containing only path parameters
"""
return {k: v for k, v in params.items()
if k in self.parameter_type_mapping and self.parameter_type_mapping[k] == 'path'}
def get_query_parameters(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract query parameters from the combined parameters.
Args:
params: Combined parameters dictionary
Returns:
Dictionary containing only query parameters
"""
return {k: v for k, v in params.items()
if k in self.parameter_type_mapping and self.parameter_type_mapping[k] == 'query'}
def get_form_parameters(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract form parameters from the combined parameters.
Args:
params: Combined parameters dictionary
Returns:
Dictionary containing only form parameters
"""
return {k: v for k, v in params.items()
if k in self.parameter_type_mapping and self.parameter_type_mapping[k] == 'form'}
def get_request_body(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract request body parameters from the combined parameters.
Args:
params: Combined parameters dictionary
Returns:
Dictionary containing only request body parameters
"""
# Parameters that are not path, query, or form are considered request body parameters
return {k: v for k, v in params.items()
if k not in self.parameter_type_mapping or self.parameter_type_mapping[k] == 'body'}
def create_simple_endpoint(endpoint: Endpoint) -> SimpleEndpoint:
"""
Transform a regular Endpoint into a SimpleEndpoint with combined parameters.
Args:
endpoint: The original Endpoint object
Returns:
A new SimpleEndpoint object with combined parameter schema
"""
# Create a parameter mapping to track where each parameter belongs
parameter_type_mapping = {}
# Start building the combined schema
combined_schema = {
'type': 'object',
'properties': {},
'required': []
}
# Add path parameters to the combined schema
if endpoint.path_parameters_schema and 'properties' in endpoint.path_parameters_schema:
for param_name, param_schema in endpoint.path_parameters_schema['properties'].items():
combined_schema['properties'][param_name] = param_schema
parameter_type_mapping[param_name] = 'path'
# Add required path parameters
if 'required' in endpoint.path_parameters_schema:
combined_schema['required'].extend(endpoint.path_parameters_schema['required'])
# Add query parameters to the combined schema
if endpoint.query_parameters_schema and 'properties' in endpoint.query_parameters_schema:
for param_name, param_schema in endpoint.query_parameters_schema['properties'].items():
combined_schema['properties'][param_name] = param_schema
parameter_type_mapping[param_name] = 'query'
# Add required query parameters
if 'required' in endpoint.query_parameters_schema:
combined_schema['required'].extend(endpoint.query_parameters_schema['required'])
# Add form parameters to the combined schema
if endpoint.form_parameters_schema and 'properties' in endpoint.form_parameters_schema:
for param_name, param_schema in endpoint.form_parameters_schema['properties'].items():
combined_schema['properties'][param_name] = param_schema
parameter_type_mapping[param_name] = 'form'
# Add required form parameters
if 'required' in endpoint.form_parameters_schema:
combined_schema['required'].extend(endpoint.form_parameters_schema['required'])
# Add request body parameters to the combined schema
if endpoint.request_body_schema:
# For simplicity, we'll assume JSON content type and look for a schema
if 'properties' in endpoint.request_body_schema:
for param_name, param_schema in endpoint.request_body_schema['properties'].items():
combined_schema['properties'][param_name] = param_schema
parameter_type_mapping[param_name] = 'body'
# Add required body parameters
if 'required' in endpoint.request_body_schema:
combined_schema['required'].extend(endpoint.request_body_schema['required'])
# Create the SimpleEndpoint with the combined schema
return SimpleEndpoint(
path=endpoint.path,
method=endpoint.method,
operation_id=endpoint.operation_id,
summary=endpoint.summary,
description=endpoint.description,
deprecated=endpoint.deprecated,
servers=endpoint.servers,
requires_bearer_auth=endpoint.requires_bearer_auth,
requires_oauth=endpoint.requires_oauth,
oauth_scopes=endpoint.oauth_scopes,
security_requirements=endpoint.security_requirements,
combined_parameter_schema=combined_schema,
parameter_type_mapping=parameter_type_mapping,
request_content_types=endpoint.request_content_types,
response_content_types=endpoint.response_content_types,
responses=endpoint.responses,
tags=endpoint.tags
)
