cBioPortal MCP Server

""" Base endpoint class for the cBioPortal MCP server. This module provides a base class that contains common functionality shared across all endpoint classes, including pagination logic, error handling, and response formatting. """ import asyncio from functools import wraps from typing import Any, Dict, List, Optional, Union from ..api_client import APIClient from ..constants import FETCH_ALL_PAGE_SIZE from ..utils.logging import get_logger from ..utils.pagination import collect_all_results from ..utils.validation import validate_page_params, validate_sort_params logger = get_logger(__name__) def handle_api_errors(operation_name: str): """ Decorator to handle common API errors consistently across endpoints. Args: operation_name: Description of the operation being performed """ def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): try: # Ensure API client is ready for BaseEndpoint instances if len(args) > 0 and hasattr(args[0], '_ensure_api_client_ready'): await args[0]._ensure_api_client_ready() return await func(*args, **kwargs) except (ValueError, TypeError) as e: # Re-raise validation errors so they can be caught by tests raise e except Exception as e: logger.error(f"Error in {operation_name}: {str(e)}") return {"error": f"Failed to {operation_name}: {str(e)}"} return wrapper return decorator def validate_paginated_params(func): """ Decorator to validate common pagination parameters. Automatically validates page_number, page_size, limit, sort_by, and direction parameters if they exist in the function signature. """ @wraps(func) async def wrapper(self, *args, **kwargs): # Extract parameters from positional args based on function signature import inspect sig = inspect.signature(func) param_names = list(sig.parameters.keys())[1:] # Skip 'self' # Build a dictionary of parameter values bound_args = sig.bind(self, *args, **kwargs) bound_args.apply_defaults() page_number = bound_args.arguments.get('page_number', 0) page_size = bound_args.arguments.get('page_size', 50) limit = bound_args.arguments.get('limit', None) sort_by = bound_args.arguments.get('sort_by', None) direction = bound_args.arguments.get('direction', 'ASC') # Validate pagination parameters - let exceptions bubble up validate_page_params(page_number, page_size, limit) validate_sort_params(sort_by, direction) return await func(self, *args, **kwargs) return wrapper class BaseEndpoint: """ Base class for all endpoint classes. Provides common functionality including pagination logic, error handling, and response formatting. """ def __init__(self, api_client: APIClient): self.api_client = api_client async def _ensure_api_client_ready(self): """Ensure APIClient is initialized before making requests.""" if not hasattr(self.api_client, '_client') or self.api_client._client is None: await self.api_client.startup() logger.info("APIClient initialized via BaseEndpoint._ensure_api_client_ready") def build_pagination_params( self, page_number: int, page_size: int, sort_by: Optional[str], direction: str, limit: Optional[int] = None ) -> Dict[str, Any]: """ Build standardized pagination parameters for API requests. Args: page_number: Page number to retrieve (0-based) page_size: Number of items per page sort_by: Field to sort by direction: Sort direction (ASC or DESC) limit: Maximum number of items to return Returns: Dictionary of API parameters """ params = { "pageNumber": page_number, "pageSize": page_size, "direction": direction, } if sort_by: params["sortBy"] = sort_by if limit == 0: params["pageSize"] = FETCH_ALL_PAGE_SIZE return params def build_pagination_response( self, results: List[Dict[str, Any]], page_number: int, page_size: int, has_more: bool, data_key: str ) -> Dict[str, Any]: """ Build standardized pagination response structure. Args: results: List of result items page_number: Current page number page_size: Page size used has_more: Whether more results are available data_key: Key name for the results in the response Returns: Standardized response dictionary """ return { data_key: results, "pagination": { "page": page_number, "page_size": page_size, "total_found": len(results), "has_more": has_more, }, } def determine_has_more( self, results: List[Dict[str, Any]], api_params: Dict[str, Any] ) -> bool: """ Determine if more results are available based on the API response. Args: results: Results returned from the API api_params: Parameters used in the API call Returns: True if more results might be available, False otherwise """ api_might_have_more = len(results) == api_params["pageSize"] # If 'fetch all' was intended and API returned less than max fetch size, # then it's definitely the end if (api_params["pageSize"] == FETCH_ALL_PAGE_SIZE and len(results) < FETCH_ALL_PAGE_SIZE): api_might_have_more = False return api_might_have_more def apply_limit( self, results: List[Dict[str, Any]], limit: Optional[int] ) -> List[Dict[str, Any]]: """ Apply limit to results if specified. Args: results: List of results to limit limit: Maximum number of results to return Returns: Limited results list """ if limit and limit > 0 and len(results) > limit: return results[:limit] return results async def paginated_request( self, endpoint: str, page_number: int = 0, page_size: int = 50, sort_by: Optional[str] = None, direction: str = "ASC", limit: Optional[int] = None, data_key: str = "results", additional_params: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Make a paginated API request with standardized handling. Args: endpoint: API endpoint to call page_number: Page number to retrieve (0-based) page_size: Number of items per page sort_by: Field to sort by direction: Sort direction (ASC or DESC) limit: Maximum number of items to return data_key: Key name for results in response additional_params: Additional parameters to include in the request Returns: Standardized paginated response """ # Build base pagination parameters api_params = self.build_pagination_params( page_number, page_size, sort_by, direction, limit ) # Add any additional parameters if additional_params: api_params.update(additional_params) # Special behavior for limit=0 (fetch all results) if limit == 0: results_from_api = await collect_all_results( self.api_client, endpoint, params=api_params ) results_for_response = results_from_api has_more = False # We fetched everything else: # Fetch just the requested page results_from_api = await self.api_client.make_api_request( endpoint, params=api_params ) # Apply limit if specified results_for_response = self.apply_limit(results_from_api, limit) # Determine if there might be more data available has_more = self.determine_has_more(results_from_api, api_params) return self.build_pagination_response( results_for_response, page_number, page_size, has_more, data_key ) async def concurrent_fetch( self, fetch_tasks: List[asyncio.Task], operation_name: str = "concurrent operation" ) -> Dict[str, Any]: """ Execute multiple API requests concurrently and return structured results. Args: fetch_tasks: List of async tasks to execute operation_name: Description of the operation for logging Returns: Dictionary with results and metadata """ import time start_time = time.perf_counter() results = await asyncio.gather(*fetch_tasks) end_time = time.perf_counter() # Process results success_count = sum(1 for r in results if r.get("success", True)) error_count = len(results) - success_count return { "results": results, "metadata": { "count": len(results), "errors": error_count, "execution_time": round(end_time - start_time, 3), "concurrent": True, "operation": operation_name, }, }