openapi-mcp-proxy

Instructions

Implementation Reference

• openapi_mcp_proxy/tools/api_exploration.py:56-71 (handler)

  The handle_call method of ListEndpointsTool that executes the core tool logic: validates API identifier, extracts pagination and filters, calls the explorer service for paginated endpoints, formats the output, and returns the MCP TextContent response.

  async def handle_call(self, arguments: Dict[str, Any]) -> List[TextContent]:
      try:
          self._validate_api_identifier(arguments["api"])
  
          pagination = self.extract_pagination_params(arguments)
          filters = self.extract_endpoint_filter_params(arguments)
  
          paginated_result = await self.explorer.list_endpoints_paginated(
              arguments["api"], pagination, filters
          )
  
          result = self._format_paginated_endpoint_response(paginated_result, filters)
          return self._create_text_response(result)
      except Exception as e:
          return self._create_error_response(e)

• openapi_mcp_proxy/tools/base.py:256-263 (schema)

  Static method defining the JSON schema for list_endpoints tool input, including required 'api' identifier, optional pagination (page, page_size), and endpoint filters (methods, tags_include/exclude, has_authentication, deprecated).

  def create_paginated_endpoint_input_schema() -> Dict[str, Any]:
      """Create input schema for paginated endpoint operations."""
      schema = ToolDefinitionMixin.create_api_input_schema()
      schema["properties"].update(ToolDefinitionMixin.create_pagination_properties())
      schema["properties"].update(
          ToolDefinitionMixin.create_endpoint_filter_properties()
      )
      return schema

• openapi_mcp_proxy/services/tool_registry.py:39-51 (registration)

  Instantiation of ListEndpointsTool with config_manager and explorer dependencies, added to the tools list in ToolRegistry._register_tools for MCP server registration.

  tools = [
      # API Management Tools
      AddApiTool(self.config_manager),
      ListSavedApisTool(self.config_manager),
      RemoveApiTool(self.config_manager),
      # API Exploration Tools
      GetApiInfoTool(self.config_manager, self.explorer),
      ListEndpointsTool(self.config_manager, self.explorer),
      SearchEndpointsTool(self.config_manager, self.explorer),
      GetEndpointDetailsTool(self.config_manager, self.explorer),
      ListModelsTool(self.config_manager, self.explorer),
      GetModelSchemaTool(self.config_manager, self.explorer),
  ]

• openapi_mcp_proxy/services/openapi_explorer.py:174-200 (helper)

  Helper service method implementing pagination and optional filtering on top of list_endpoints, directly called by the tool handler.

  async def list_endpoints_paginated(
      self,
      api_identifier: str,
      pagination: PaginationParams,
      filters: Optional[EndpointFilterParams] = None,
  ) -> PaginationResult[EndpointInfo]:
      """List endpoints with pagination and filtering."""
      all_endpoints = await self.list_endpoints(api_identifier)
  
      if filters:
          filtered_endpoints = [
              ep for ep in all_endpoints if ep.matches_filters(filters)
          ]
      else:
          filtered_endpoints = all_endpoints
  
      total_count = len(filtered_endpoints)
      start_idx = pagination.get_offset()
      end_idx = start_idx + pagination.get_limit()
      paginated_endpoints = filtered_endpoints[start_idx:end_idx]
  
      logger.info(
          f"Paginated endpoints for API {api_identifier}: "
          f"page {pagination.page}, showing {len(paginated_endpoints)} of {total_count}"
      )
  
      return PaginationResult.create(paginated_endpoints, total_count, pagination)

• openapi_mcp_proxy/services/openapi_explorer.py:44-74 (helper)

  Core helper method that parses the OpenAPI schema to extract and construct EndpointInfo objects for all paths and valid HTTP methods, used by paginated version.

  async def list_endpoints(self, api_identifier: str) -> List[EndpointInfo]:
      """List all endpoints in an API."""
      url, headers = self.config_manager.get_api_config(api_identifier)
      schema = await self.cache.get_schema(url, headers)
  
      endpoints = []
      paths = schema.get("paths", {})
  
      for path, path_info in paths.items():
          for method, operation in path_info.items():
              if self._is_valid_http_method(method):
                  deprecated = operation.get("deprecated", False)
  
                  has_auth = bool(operation.get("security", [])) or bool(
                      schema.get("security", [])
                  )
  
                  endpoint = EndpointInfo(
                      path=path,
                      method=method.upper(),
                      summary=operation.get("summary"),
                      description=operation.get("description"),
                      tags=operation.get("tags", []),
                      operation_id=operation.get("operationId"),
                      deprecated=deprecated,
                      has_authentication=has_auth,
                  )
                  endpoints.append(endpoint)
  
      logger.info(f"Found {len(endpoints)} endpoints for API {api_identifier}")
      return endpoints